Storage - SeekableByteChannel Unnecessary Request Bugfix

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

@@ -7,6 +7,14 @@
 ### Breaking Changes
 
 ### Bugs Fixed
+- Fixed an issue where `BlobClientBase.openSeekableByteChannelRead` issued an unnecessary HTTP request (resulting
+  in an HTTP 416 response) after the entire blob had already been returned in the initial range download. When the
+  channel is opened with ETag consistency control (the default), the read behavior now short-circuits to end-of-file
+  once the known resource length is reached, avoiding the extra round trip.
+- Fixed an issue where a transport-level failure while streaming the body of the trailing HTTP 416 response from
+  `BlobClientBase.openSeekableByteChannelRead` (for example "Connection reset by peer") could propagate out of the
+  channel even though all of the blob's content had already been delivered to the caller. The read behavior now
+  logs a warning and signals end-of-file when such an error occurs at or past the known end of the resource.
 
 ### Other Changes
 

sdk/storage/azure-storage-blob/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "java",
   "TagPrefix": "java/storage/azure-storage-blob",
-  "Tag": "java/storage/azure-storage-blob_47f4243e59"
+  "Tag": "java/storage/azure-storage-blob_552df730ec"
 }

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

@@ -71,6 +71,15 @@ public int read(ByteBuffer dst, long sourceOffset) throws IOException {
             initialBufferPosition = null;
         }
 
+        // If the request is at or past the known end of the resource and the blob content is locked via an
+        // If-Match ETag, the blob cannot have grown without invalidating the precondition. Short-circuit and
+        // signal end-of-file rather than issuing a request that the service will reject with HTTP 416. This
+        // avoids an unnecessary round trip (and the failure modes that come with streaming the 416 response
+        // body, such as the connection being reset by the service).
+        if (sourceOffset >= resourceLength && isEtagLocked()) {
+            return -1;
+        }
+
         int initialPosition = dst.position();
 
         try (ByteBufferBackedOutputStreamUtil dstStream = new ByteBufferBackedOutputStreamUtil(dst)) {
@@ -90,7 +99,31 @@ public int read(ByteBuffer dst, long sourceOffset) throws IOException {
                 return sourceOffset < resourceLength ? 0 : -1;
             }
             throw LOGGER.logExceptionAsError(e);
+        } catch (RuntimeException e) {
+            // Reading the body of an HTTP 416 response can fail with transport-level errors (for example
+            // 'Connection reset by peer' wrapped in reactor's ReactiveException). When the requested offset is
+            // already at or past the known end of the resource, no data could have been returned anyway, so log
+            // the failure and signal end-of-file instead of propagating an exception to the caller that has
+            // already received all of the blob's content.
+            if (sourceOffset >= resourceLength) {
+                LOGGER.warning("Ignoring error encountered while issuing a read at or past the end of the blob; "
+                    + "treating as end-of-file because the resource length is already known.", e);
+                return -1;
+            }
+            throw LOGGER.logExceptionAsError(e);
+        }
+    }
+
+    /**
+     * @return Whether the request conditions on this behavior lock the blob's content via an If-Match ETag.
+     */
+    private boolean isEtagLocked() {
+        if (requestConditions == null) {
+            return false;
         }
+
+        String ifMatch = requestConditions.getIfMatch();
+        return !CoreUtils.isNullOrEmpty(ifMatch) && !"*".equals(ifMatch.trim());
     }
 
     /**

sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/specialized/StorageSeekableByteChannelBlobReadBehaviorTests.java

@@ -35,9 +35,11 @@
 import java.util.stream.Stream;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
@@ -273,4 +275,110 @@ void readDetectsBlobGrowth() throws IOException {
         assertEquals(buffer.capacity(), buffer.position());
         TestUtils.assertArraysEqual(data, halfLength, buffer.array(), 0, data.length - halfLength);
     }
+
+    /**
+     * When the request conditions lock the blob via an If-Match ETag, a read at or past the known end of the
+     * resource must short-circuit to EOF without issuing a service call (which the service would reject with an
+     * HTTP 416 response).
+     */
+    @Test
+    public void readPastEndShortCircuitsWhenETagLocked() throws IOException {
+        BlobClientBase client = Mockito.mock(BlobClientBase.class);
+
+        long resourceLength = Constants.KB;
+        BlobRequestConditions conditions = new BlobRequestConditions().setIfMatch("0xETAG");
+        StorageSeekableByteChannelBlobReadBehavior behavior = new StorageSeekableByteChannelBlobReadBehavior(client,
+            ByteBuffer.allocate(0), -1, resourceLength, conditions);
+
+        // when: "Reading at the known end of the resource"
+        int readAtEnd = behavior.read(ByteBuffer.allocate(Constants.KB), resourceLength);
+
+        // then: "EOF is signaled without any service call"
+        assertEquals(-1, readAtEnd);
+        verify(client, never()).downloadStreamWithResponse(any(), any(), any(), any(), anyBoolean(), any(), any());
+
+        // when: "Reading past the known end of the resource"
+        int readPastEnd = behavior.read(ByteBuffer.allocate(Constants.KB), resourceLength + Constants.KB);
+
+        // then: "EOF is still signaled without any service call"
+        assertEquals(-1, readPastEnd);
+        verify(client, never()).downloadStreamWithResponse(any(), any(), any(), any(), anyBoolean(), any(), any());
+
+        // If-Match="*" is an existence precondition (not a specific ETag lock), so growth-detection behavior should be
+        // preserved and a request should still be issued.
+        Mockito.when(client.downloadStreamWithResponse(any(), any(), any(), any(), anyBoolean(), any(), any()))
+            .thenReturn(createMockDownloadResponse(
+                "bytes " + resourceLength + "-" + (resourceLength + Constants.KB - 1) + "/" + 2 * Constants.KB));
+
+        BlobRequestConditions ifMatchStar = new BlobRequestConditions().setIfMatch("*");
+        StorageSeekableByteChannelBlobReadBehavior starBehavior = new StorageSeekableByteChannelBlobReadBehavior(client,
+            ByteBuffer.allocate(0), -1, resourceLength, ifMatchStar);
+
+        starBehavior.read(ByteBuffer.allocate(Constants.KB), resourceLength);
+
+        verify(client, times(1)).downloadStreamWithResponse(any(), any(), any(), any(), anyBoolean(), any(), any());
+    }
+
+    /**
+     * When the blob is not ETag-locked, a read past the end must still issue a request so the behavior can
+     * detect blob growth (existing contract preserved).
+     */
+    @Test
+    public void readPastEndIssuesRequestWhenNotETagLocked() throws IOException {
+        BlobClientBase client = Mockito.mock(BlobClientBase.class);
+        long resourceLength = Constants.KB;
+        Mockito.when(client.downloadStreamWithResponse(any(), any(), any(), any(), anyBoolean(), any(), any()))
+            .thenReturn(createMockDownloadResponse(
+                "bytes " + resourceLength + "-" + (resourceLength + Constants.KB - 1) + "/" + 2 * Constants.KB));
+
+        StorageSeekableByteChannelBlobReadBehavior behavior
+            = new StorageSeekableByteChannelBlobReadBehavior(client, ByteBuffer.allocate(0), -1, resourceLength, null);
+
+        // when: "Reading past the known end of the resource without an ETag lock"
+        behavior.read(ByteBuffer.allocate(Constants.KB), resourceLength);
+
+        // then: "A service call is issued (existing growth-detection behavior preserved)"
+        verify(client, times(1)).downloadStreamWithResponse(any(), any(), any(), any(), anyBoolean(), any(), any());
+    }
+
+    /**
+     * If the underlying download throws a non-{@code BlobStorageException} (for example, the connection is
+     * reset while streaming the body of a 416 response) and the caller is already at or past the known end of
+     * the resource, the behavior should swallow the error, log a warning, and signal EOF rather than throwing.
+     */
+    @Test
+    public void readPastEndSwallowsTransportErrorAndSignalsEof() throws IOException {
+        BlobClientBase client = Mockito.mock(BlobClientBase.class);
+        long resourceLength = Constants.KB;
+        Mockito.when(client.downloadStreamWithResponse(any(), any(), any(), any(), anyBoolean(), any(), any()))
+            .thenThrow(new RuntimeException("Connection reset by peer"));
+
+        StorageSeekableByteChannelBlobReadBehavior behavior
+            = new StorageSeekableByteChannelBlobReadBehavior(client, ByteBuffer.allocate(0), -1, resourceLength, null);
+
+        // when: "Reading past the known end and the download fails with a transport-level error"
+        int read = behavior.read(ByteBuffer.allocate(Constants.KB), resourceLength);
+
+        // then: "EOF is signaled instead of the exception propagating"
+        assertEquals(-1, read);
+        verify(client, times(1)).downloadStreamWithResponse(any(), any(), any(), any(), anyBoolean(), any(), any());
+    }
+
+    /**
+     * Non-{@code BlobStorageException} failures that occur while reading within the known bounds of the resource
+     * must continue to surface as exceptions, since some bytes the caller asked for could not be retrieved.
+     */
+    @Test
+    public void readWithinResourcePropagatesTransportError() {
+        BlobClientBase client = Mockito.mock(BlobClientBase.class);
+        long resourceLength = Constants.KB;
+        Mockito.when(client.downloadStreamWithResponse(any(), any(), any(), any(), anyBoolean(), any(), any()))
+            .thenThrow(new RuntimeException("Connection reset by peer"));
+
+        StorageSeekableByteChannelBlobReadBehavior behavior
+            = new StorageSeekableByteChannelBlobReadBehavior(client, ByteBuffer.allocate(0), -1, resourceLength, null);
+
+        // Reading within the known resource range should still surface the error.
+        assertThrows(RuntimeException.class, () -> behavior.read(ByteBuffer.allocate(Constants.KB), 0));
+    }
 }