HADOOP-16785: wasb to raise IOE if write() invoked on a closed stream

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/LambdaTestUtils.java

@@ -406,7 +406,7 @@ public static <E extends Throwable> E intercept(
       throws Exception {
     try {
       eval.call();
-      throw new AssertionError("Expected an exception");
+      throw new AssertionError("Expected an exception of type " + clazz);
     } catch (Throwable e) {
       if (clazz.isAssignableFrom(e.getClass())) {
         return (E)e;

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/NativeAzureFileSystem.java

@@ -1083,6 +1083,7 @@ public synchronized void close() throws IOException {
      */
     @Override
     public void write(int b) throws IOException {
+      checkOpen();
       try {
         out.write(b);
       } catch(IOException e) {
@@ -1106,6 +1107,7 @@ public void write(int b) throws IOException {
      */
     @Override
     public void write(byte[] b) throws IOException {
+      checkOpen();
       try {
         out.write(b);
       } catch(IOException e) {
@@ -1136,6 +1138,7 @@ public void write(byte[] b) throws IOException {
      */
     @Override
     public void write(byte[] b, int off, int len) throws IOException {
+      checkOpen();
       try {
         out.write(b, off, len);
       } catch(IOException e) {
@@ -1198,6 +1201,17 @@ public void setEncodedKey(String anEncodedKey) {
     private void restoreKey() throws IOException {
       store.rename(getEncodedKey(), getKey());
     }
+
+    /**
+     * Check for the stream being open.
+     * @throws IOException if the stream is closed.
+     */
+    private void checkOpen() throws IOException {
+      if (out == null) {
+        throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
+      }
+    }
+
   }
 
   private URI uri;

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStream.java

@@ -259,7 +259,15 @@ public synchronized void close() throws IOException {
   }
 
   private synchronized void flushInternal(boolean isClose) throws IOException {
-    maybeThrowLastError();
+    try {
+      maybeThrowLastError();
+    } catch (IOException e) {
+      if (isClose) {
+        // wrap existing exception so as to avoid breaking try-with-resources
+        throw new IOException("Skipping final flush and write due to " + e, e);
+      } else
+        throw e;
+    }
     writeCurrentBufferToService();
     flushWrittenBytesToService(isClose);
   }

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/AbstractWasbTestBase.java

@@ -52,7 +52,7 @@ public abstract class AbstractWasbTestBase extends AbstractWasbTestWithTimeout
   @Before
   public void setUp() throws Exception {
     AzureBlobStorageTestAccount account = createTestAccount();
-    assumeNotNull(account);
+    assumeNotNull("test account", account);
     bindToTestAccount(account);
   }
 

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/ITestFileSystemOperationExceptionHandling.java

@@ -19,6 +19,7 @@
 package org.apache.hadoop.fs.azure;
 
 import java.io.FileNotFoundException;
+import java.io.IOException;
 
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
@@ -30,7 +31,9 @@
 import org.junit.After;
 import org.junit.Test;
 
+import static org.apache.hadoop.fs.FSExceptionMessages.STREAM_IS_CLOSED;
 import static org.apache.hadoop.fs.azure.ExceptionHandlingTestHelper.*;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 
 /**
  * Single threaded exception handling.
@@ -265,6 +268,23 @@ public void testSingleThreadedPageBlobOpenScenario() throws Throwable {
     inputStream = fs.open(testPath);
   }
 
+  /**
+   * Attempts to write to the azure stream after it is closed will raise
+   * an IOException.
+   */
+  @Test
+  public void testWriteAfterClose() throws Throwable {
+    FSDataOutputStream out = fs.create(testPath);
+    out.close();
+    intercept(IOException.class, STREAM_IS_CLOSED,
+        () -> out.write('a'));
+    intercept(IOException.class, STREAM_IS_CLOSED,
+        () -> out.write(new byte[]{'a'}));
+    out.hsync();
+    out.flush();
+    out.close();
+  }
+
   @After
   public void tearDown() throws Exception {
     if (inputStream != null) {

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemCreate.java

@@ -19,6 +19,8 @@
 package org.apache.hadoop.fs.azurebfs;
 
 import java.io.FileNotFoundException;
+import java.io.FilterOutputStream;
+import java.io.IOException;
 import java.util.EnumSet;
 
 import org.junit.Test;
@@ -29,6 +31,7 @@
 import org.apache.hadoop.fs.permission.FsPermission;
 
 import static org.apache.hadoop.fs.contract.ContractTestUtils.assertIsFile;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 
 /**
  * Test create operation.
@@ -104,4 +107,39 @@ public void testCreateNonRecursive2() throws Exception {
         .close();
     assertIsFile(fs, testFile);
   }
+
+  /**
+   * Attempts to use to the ABFS stream after it is closed.
+   */
+  @Test
+  public void testWriteAfterClose() throws Throwable {
+    final AzureBlobFileSystem fs = getFileSystem();
+    Path testPath = new Path(TEST_FOLDER_PATH, TEST_CHILD_FILE);
+    FSDataOutputStream out = fs.create(testPath);
+    out.close();
+    intercept(IOException.class, () -> out.write('a'));
+    intercept(IOException.class, () -> out.write(new byte[]{'a'}));
+    // hsync is not ignored on a closed stream
+    // out.hsync();
+    out.flush();
+    out.close();
+  }
+
+  /**
+   * Attempts to double close an ABFS output stream from within a
+   * FilterOutputStream.
+   * That class handles a double failure on close badly if the second
+   * exception rethrows the first.
+   */
+  @Test
+  public void testFilteredDoubleClose() throws Throwable {
+    final AzureBlobFileSystem fs = getFileSystem();
+    Path testPath = new Path(TEST_FOLDER_PATH, TEST_CHILD_FILE);
+    FilterOutputStream out = new FilterOutputStream(fs.create(testPath));
+    out.close();
+    intercept(IOException.class, () -> out.write('a'));
+    intercept(IOException.class, () -> out.write(new byte[]{'a'}));
+    out.flush();
+    out.close();
+  }
 }