Added support to lock on version in BlobInputStream (Azure#17908)

Author: gapra-msft

sdk/storage/azure-storage-blob/CHANGELOG.md

@@ -1,13 +1,13 @@
 # Release History
 
 ## 12.11.0-beta.1 (Unreleased)
+- Added support to lock on version id by specifying a consistent read control when opening a BlobInputStream.
 - Removed a deep copy in the general upload path to reduce memory consumption and increase perf
 - Added a deep copy immediately after calling BlobOutputStream.write to prevent overwriting data in the case of reusing a single buffer to write to an output stream
 
 ## 12.10.0 (2021-01-14)
 - GA release
 
-
 ## 12.10.0-beta.1 (2020-12-07)
 - Exposed ClientOptions on all client builders, allowing users to set a custom application id and custom headers.
 - Added ability to get container client from blob clients and service client from container clients

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/models/ConsistentReadControl.java

@@ -0,0 +1,40 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.storage.blob.models;
+
+import com.azure.storage.blob.BlobClient;
+
+/**
+ * Defines values to indicate what strategy the SDK should use when reading from a blob to ensure the view of the data
+ * is consistent and not changed during the read.
+ * {@link #NONE}
+ * {@link #ETAG}
+ * {@link #VERSION_ID}
+ */
+public enum ConsistentReadControl {
+    /**
+     * No consistent read control. The client will honor user provided {@link BlobRequestConditions#getIfMatch()} and
+     * {@link BlobClient#getVersionId()}.
+     */
+    NONE,
+
+    /**
+     * Default value. Consistent read control based on eTag.
+     * If {@link BlobRequestConditions#getIfMatch()} is set, the client will honor this value.
+     * Otherwise, {@link BlobRequestConditions#getIfMatch()} is set to the latest eTag.
+     * Note: Modification of the base blob will result in an {@code IOException} or a {@code BlobStorageException} if
+     * eTag is the only form of consistent read control being employed.
+     */
+    ETAG,
+
+    /**
+     * Consistent control based on versionId. Note: Versioning must be supported by the account to use this value.
+     * If {@link BlobClient#getVersionId()} is set, the client will honor this value.
+     * Otherwise, {@link BlobClient#getVersionId()} is set to the latest versionId.
+     * Note: Modification of the base blob will not result in an {@code Exception} and allow you to continue reading the
+     * entirety of the appropriate version of the blob determined at the time of opening the {@code InputStream} but it
+     * may no longer be the latest data.
+     */
+    VERSION_ID
+}

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/options/BlobInputStreamOptions.java

@@ -6,6 +6,7 @@
 import com.azure.core.annotation.Fluent;
 import com.azure.storage.blob.models.BlobRange;
 import com.azure.storage.blob.models.BlobRequestConditions;
+import com.azure.storage.blob.models.ConsistentReadControl;
 
 /**
  * Extended options that may be passed when opening a blob input stream.
@@ -16,6 +17,7 @@ public class BlobInputStreamOptions {
     private BlobRange range;
     private BlobRequestConditions requestConditions;
     private Integer blockSize;
+    private ConsistentReadControl consistentReadControl;
 
     /**
      * @return {@link BlobRange}
@@ -68,4 +70,20 @@ public BlobInputStreamOptions setBlockSize(Integer blockSize) {
         this.blockSize = blockSize;
         return this;
     }
+
+    /**
+     * @return {@link ConsistentReadControl} Default is E-Tag.
+     */
+    public ConsistentReadControl getConsistentReadControl() {
+        return consistentReadControl;
+    }
+
+    /**
+     * @param consistentReadControl {@link ConsistentReadControl} Default is E-Tag.
+     * @return The updated options.
+     */
+    public BlobInputStreamOptions setConsistentReadControl(ConsistentReadControl consistentReadControl) {
+        this.consistentReadControl = consistentReadControl;
+        return this;
+    }
 }

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobClientBase.java

@@ -15,6 +15,7 @@
 import com.azure.storage.blob.BlobContainerClient;
 import com.azure.storage.blob.BlobServiceClient;
 import com.azure.storage.blob.implementation.util.ModelHelper;
+import com.azure.storage.blob.models.ConsistentReadControl;
 import com.azure.storage.blob.options.BlobBeginCopyOptions;
 import com.azure.storage.blob.options.BlobCopyFromUrlOptions;
 import com.azure.storage.blob.models.BlobProperties;
@@ -261,17 +262,43 @@ public final BlobInputStream openInputStream(BlobRange range, BlobRequestConditi
      */
     public BlobInputStream openInputStream(BlobInputStreamOptions options) {
         options = options == null ? new BlobInputStreamOptions() : options;
+        ConsistentReadControl consistentReadControl = options.getConsistentReadControl() == null
+            ? ConsistentReadControl.ETAG : options.getConsistentReadControl();
 
-        BlobProperties properties = getProperties();
+        BlobProperties properties = getPropertiesWithResponse(options.getRequestConditions(), null, null).getValue();
+        String eTag = properties.getETag();
+        String versionId = properties.getVersionId();
 
         BlobRange range = options.getRange() == null ? new BlobRange(0) : options.getRange();
         int chunkSize = options.getBlockSize() == null ? 4 * Constants.MB : options.getBlockSize();
 
         BlobRequestConditions requestConditions = options.getRequestConditions() == null
             ? new BlobRequestConditions() : options.getRequestConditions();
-        // Target the user specified version by default. If not provided, target the latest version.
-        if (requestConditions.getIfMatch() == null) {
-            requestConditions.setIfMatch(properties.getETag());
+        BlobAsyncClientBase client = this.client;
+
+        switch (consistentReadControl) {
+            case NONE:
+                break;
+            case ETAG:
+                // Target the user specified eTag by default. If not provided, target the latest eTag.
+                if (requestConditions.getIfMatch() == null) {
+                    requestConditions.setIfMatch(eTag);
+                }
+                break;
+            case VERSION_ID:
+                if (versionId == null) {
+                    throw logger.logExceptionAsError(
+                        new UnsupportedOperationException("Versioning is not supported on this account."));
+                } else {
+                    // Target the user specified version by default. If not provided, target the latest version.
+                    if (this.client.getVersionId() == null) {
+                        client = this.client.getVersionClient(versionId);
+                    }
+                }
+                break;
+            default:
+                throw logger.logExceptionAsError(new IllegalArgumentException("Concurrency control type not "
+                    + "supported."));
         }
 
         return new BlobInputStream(client, range.getOffset(), range.getCount(), chunkSize,