Added abort and close

Author: rickle-msft

ChangeLog.txt

@@ -1,3 +1,6 @@
+2020.03.25 Version 8.6.3
+ * Added the commitWriteOnInputStreamException option to BlobRequestOptions, which will allow the user to configure whether any data staged through openWrite when using the upload(InputStream, long) method will be committed upon failures in the InputStream.
+
 2020.03.18 Version 8.6.2
  * Fixed a bug in the pom that disrupted the ability to download from maven central.
 

microsoft-azure-storage-test/src/com/microsoft/azure/storage/blob/CloudBlockBlobTests.java

@@ -2755,6 +2755,71 @@ public void testSkipEtagCheck() throws StorageException, IOException, URISyntaxE
         stream.close();
     }
 
+    private static class ExceptionInputStream extends InputStream {
+        private final byte[] data;
+        int index = 0;
+        boolean firstRead = true;
+
+        ExceptionInputStream(byte[] data) {
+            this.data = data;
+        }
+
+        @Override
+        public int read() throws IOException {
+            return 0;
+        }
+
+        @Override
+        public int read(byte[] arr, int offset, int len) throws IOException {
+            if (firstRead) {
+                // Fill either half the incoming buffer or use half the data, whichever is smaller.
+                // For safe partial write.
+                int size = Math.min(data.length, len) / 2;
+                if (len >= 0) System.arraycopy(data, index, arr, offset, size);
+                firstRead = false;
+                return size;
+            } else {
+                throw new IOException();
+            }
+        }
+    }
+
+    @Test
+    public void testCommitOnInputStreamException() throws StorageException, IOException, URISyntaxException {
+        final int blobSize = 2 * Constants.DEFAULT_MINIMUM_READ_SIZE_IN_BYTES; // so BlobInputStream doesn't read entire blob at once.
+
+        CloudBlobContainer container = BlobTestHelper.getRandomContainerReference();
+        container.createIfNotExists();
+        CloudBlockBlob blob = container.getBlockBlobReference(BlobTestHelper.generateRandomBlobNameWithPrefix(""));
+
+        BlobRequestOptions options = new BlobRequestOptions();
+
+        // Upload with no commit on failure.
+        byte[] data = TestHelper.getRandomBuffer(blobSize);
+        InputStream is = new ExceptionInputStream(data);
+        options.setCommitWriteOnInputStreamException(false);
+
+        try {
+            blob.upload(is, blobSize, null, null, options, null);
+            fail();
+        } catch(IOException e) {
+            assertFalse(blob.exists());
+        }
+
+        // Upload with commit on failure.
+        is = new ExceptionInputStream(data);
+        options = new BlobRequestOptions(); // Test the default. Should be true.
+
+        try {
+            blob.upload(is, blobSize, null, null, options, null);
+            fail();
+        } catch (IOException e) {
+            assertTrue(blob.exists());
+        }
+
+        container.delete();
+    }
+
     @Test
     public void testPutGetBlobCPK() throws URISyntaxException, StorageException, IOException {
         // load CPK into options

microsoft-azure-storage/src/com/microsoft/azure/storage/blob/BlobDecryptStream.java

@@ -59,6 +59,11 @@ public void close() throws IOException
         this.cryptoStream.close();
     }
 
+    @Override
+    void abortAndClose() throws IOException {
+        this.cryptoStream.close();
+    }
+
     @Override
     public void write(byte[] data, int offset, int length) throws IOException {
         // Keep buffering until we have 16 bytes of IV.

microsoft-azure-storage/src/com/microsoft/azure/storage/blob/BlobEncryptStream.java

@@ -159,4 +159,9 @@ public void close() throws IOException {
         this.cipherStream.close();
     }
 
+    @Override
+    void abortAndClose() throws IOException {
+        this.cipherStream.close();
+    }
+
 }

microsoft-azure-storage/src/com/microsoft/azure/storage/blob/BlobOutputStream.java

@@ -111,4 +111,10 @@ public void write(final byte[] data) throws IOException {
     @Override
     @DoesServiceRequest
     public abstract void close() throws IOException;
+
+    /**
+     * Closes the output stream without committing the data to the service, typically to be used in cases of errors or
+     * exceptions in the data source.
+     */
+    abstract void abortAndClose() throws IOException;
 }

microsoft-azure-storage/src/com/microsoft/azure/storage/blob/BlobOutputStreamInternal.java

@@ -339,6 +339,21 @@ public synchronized void close() throws IOException {
         }
     }
 
+    @Override
+    public void abortAndClose() throws IOException {
+        try {
+            this.checkStreamState();
+
+            this.threadExecutor.shutdown();
+        } finally {
+            this.lastError = new IOException(SR.STREAM_CLOSED);
+
+            if (!this.threadExecutor.isShutdown()) {
+                this.threadExecutor.shutdown();
+            }
+        }
+    }
+
     /**
      * Commits the blob, for block blob this uploads the block list.
      * 

microsoft-azure-storage/src/com/microsoft/azure/storage/blob/BlobRequestOptions.java

@@ -89,7 +89,15 @@ public final class BlobRequestOptions extends RequestOptions {
      * Default is false.
      */
     private boolean skipEtagLocking = false;
-    
+
+    /**
+     * A <code>boolean</code> that defines the behavior for handling exceptions when reading from the
+     * <code>InputStream</code> and using <code>openWrite</code>. If <code>true</code> the data that has been read from
+     * the stream up to the point of the exception will be flushed and a new blob will be committed with that data.
+     * Otherwise, the upload will be aborted and no data will be committed.
+     */
+     private boolean commitWriteOnInputStreamException = true;
+
     /**
      * Creates an instance of the <code>BlobRequestOptions</code> class.
      */
@@ -119,6 +127,7 @@ public BlobRequestOptions(final BlobRequestOptions other) {
             //this.setSourceCustomerProvidedKey(other.getSourceCustomerProvidedKey());
             this.setValidateEncryptionPolicy(other.getValidateEncryptionPolicy());
             this.setSkipEtagLocking(other.getSkipEtagLocking());
+            this.setCommitWriteOnInputStreamException(other.getCommitWriteOnInputStreamException());
         }
     }
 
@@ -362,6 +371,18 @@ public boolean getSkipEtagLocking() {
         return this.skipEtagLocking;
     }
 
+    /**
+     * A <code>boolean</code> that defines the behavior for handling exceptions when reading from the
+     * <code>InputStream</code> and using <code>openWrite</code>. If <code>true</code> the data that has been read from
+     * the stream up to the point of the exception will be flushed and a new blob will be committed with that data.
+     * Otherwise, the upload will be aborted and no data will be committed.
+     *
+     * @return <code>true</code> if data will be committed upon an exception; otherwise, <code>false</code>.
+     */
+    public boolean getCommitWriteOnInputStreamException() {
+        return this.commitWriteOnInputStreamException;
+    }
+
     /**
      * Sets whether a conditional failure should be absorbed on a retry attempt for the request. This option 
      * is only used by {@link CloudAppendBlob} in upload and openWrite methods. By default, it is set to 
@@ -528,6 +549,19 @@ public void setSkipEtagLocking(boolean skipEtagLocking) {
         this.skipEtagLocking = skipEtagLocking;
     }
 
+    /**
+     * A <code>boolean</code> that defines the behavior for handling exceptions when reading from the
+     * <code>InputStream</code> and using <code>openWrite</code>. If <code>true</code> the data that has been read from
+     * the stream up to the point of the exception will be flushed and a new blob will be committed with that data.
+     * Otherwise, the upload will be aborted and no data will be committed.
+     *
+     * @param commitWriteOnInputStreamException
+     *          Use <code>true</code> if data will be committed upon an exception; otherwise, <code>false</code>.
+     */
+    public void setCommitWriteOnInputStreamException(boolean commitWriteOnInputStreamException) {
+        this.commitWriteOnInputStreamException = commitWriteOnInputStreamException;
+    }
+
     /**
      * Assert that if validation is on, an encryption policy is not specified.
      */

microsoft-azure-storage/src/com/microsoft/azure/storage/blob/CloudBlockBlob.java

@@ -769,8 +769,9 @@ public void upload(final InputStream sourceStream, final long length, final Acce
      *             If a storage service error occurred.
      */
     @DoesServiceRequest
-    public void upload(final InputStream sourceStream, final long length, final StandardBlobTier standardBlobTier, final AccessCondition accessCondition,
-                       BlobRequestOptions options, OperationContext opContext) throws StorageException, IOException {
+    public void upload(final InputStream sourceStream, final long length, final StandardBlobTier standardBlobTier,
+            final AccessCondition accessCondition, BlobRequestOptions options,
+            OperationContext opContext) throws StorageException, IOException {
         if (length < -1) {
             throw new IllegalArgumentException(SR.STREAM_LENGTH_NEGATIVE);
         }
@@ -892,11 +893,20 @@ public byte[] getByteArray() {
             useOpenWrite = true;
             if (useOpenWrite) {
                 final BlobOutputStream writeStream = this.openOutputStream(accessCondition, options, opContext);
-                try {
-                    writeStream.write(inputDataStream, length);
-                }
-                finally {
-                    writeStream.close();
+                if (options.getCommitWriteOnInputStreamException()) {
+                    try {
+                        writeStream.write(inputDataStream, length);
+                    } finally {
+                        writeStream.close();
+                    }
+                } else {
+                    try {
+                        writeStream.write(inputDataStream, length);
+                        writeStream.close();
+                    } catch (Exception e) {
+                        writeStream.abortAndClose();
+                        throw e;
+                    }
                 }
             }
             else {

microsoft-azure-storage/src/com/microsoft/azure/storage/blob/LengthLimitingStream.java

@@ -81,4 +81,9 @@ public void close() throws IOException {
         // no op
     }
 
+    @Override
+    void abortAndClose() throws IOException {
+        // no op
+    }
+
 }

pom.xml

@@ -10,7 +10,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>com.microsoft.azure</groupId>
 	<artifactId>azure-storage</artifactId>
-	<version>8.6.2</version>
+	<version>8.6.3-commitOnFail</version>
 	<packaging>jar</packaging>
 
 	<name>Microsoft Azure Storage Client SDK</name>