HADOOP-19557 S3A: S3ABlockOutputStream to never log/reject hflush() calls (#7662)

S3A output streams no longer logs warnings on use of hflush()
or, if fs.s3a.downgrade.syncable.exceptions = false,
raises an UnsupportedOperationException .

hsync() is still reported with a warning or rejected. 
That method is absolutely unsupported when writing to S3.

Contributed by Steve Loughran

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ABlockOutputStream.java

@@ -829,7 +829,8 @@ class S3ABlockOutputStream extends OutputStream implements
   @Override
   public void hflush() throws IOException {
     statistics.hflushInvoked();
-    handleSyncableInvocation();
+    // do not reject these, but downgrade to a no-oop
+    LOG.debug("Hflush invoked");
   }
 
   @Override
@@ -839,7 +840,7 @@ class S3ABlockOutputStream extends OutputStream implements
   }
 
   /**
-   * Shared processing of Syncable operation reporting/downgrade.
+   * Processing of Syncable operation reporting/downgrade.
    * @throws UnsupportedOperationException if required.
    */
   private void handleSyncableInvocation() {

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/TestS3ABlockOutputStream.java

@@ -148,7 +148,7 @@ public class TestS3ABlockOutputStream extends AbstractS3AMockTest {
 
   /**
    * Unless configured to downgrade, the stream will raise exceptions on
-   * Syncable API calls.
+   * Syncable.hsync() API calls.
    */
   @Test
   public void testSyncableUnsupported() throws Exception {
@@ -156,13 +156,13 @@ public class TestS3ABlockOutputStream extends AbstractS3AMockTest {
         builder = mockS3ABuilder();
     builder.withDowngradeSyncableExceptions(false);
     stream = spy(new S3ABlockOutputStream(builder));
-    intercept(UnsupportedOperationException.class, () -> stream.hflush());
+    stream.hflush();
     intercept(UnsupportedOperationException.class, () -> stream.hsync());
   }
 
   /**
    * When configured to downgrade, the stream downgrades on
-   * Syncable API calls.
+   * Syncable.hsync() API calls.
    */
   @Test
   public void testSyncableDowngrade() throws Exception {