addressing latest review comments

Author: gunjansingh-msft

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

@@ -13,7 +13,6 @@
 import com.azure.storage.blob.implementation.accesshelpers.BlobDownloadAsyncResponseConstructorProxy;
 import com.azure.storage.blob.implementation.models.BlobsDownloadHeaders;
 import com.azure.storage.blob.implementation.util.ModelHelper;
-import com.azure.core.util.logging.ClientLogger;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -30,18 +29,8 @@
 public final class BlobDownloadAsyncResponse extends ResponseBase<BlobDownloadHeaders, Flux<ByteBuffer>>
     implements Closeable {
 
-    private static final ClientLogger LOGGER = new ClientLogger(BlobDownloadAsyncResponse.class);
-
     static {
-        BlobDownloadAsyncResponseConstructorProxy
-            .setAccessor(new BlobDownloadAsyncResponseConstructorProxy.BlobDownloadAsyncResponseConstructorAccessor() {
-                @Override
-                public BlobDownloadAsyncResponse create(StreamResponse sourceResponse,
-                    BiFunction<Throwable, Long, Mono<StreamResponse>> onErrorResume,
-                    DownloadRetryOptions retryOptions) {
-                    return new BlobDownloadAsyncResponse(sourceResponse, onErrorResume, retryOptions);
-                }
-            });
+        BlobDownloadAsyncResponseConstructorProxy.setAccessor(BlobDownloadAsyncResponse::new);
     }
 
     private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0);
@@ -67,6 +56,13 @@ public BlobDownloadAsyncResponse(HttpRequest request, int statusCode, HttpHeader
         this.retryOptions = null;
     }
 
+    /**
+     * Constructs a {@link BlobDownloadAsyncResponse}.
+     *
+     * @param sourceResponse The initial Stream Response
+     * @param onErrorResume Function used to resume.
+     * @param retryOptions Retry options.
+     */
     BlobDownloadAsyncResponse(StreamResponse sourceResponse,
         BiFunction<Throwable, Long, Mono<StreamResponse>> onErrorResume, DownloadRetryOptions retryOptions) {
         super(sourceResponse.getRequest(), sourceResponse.getStatusCode(), sourceResponse.getHeaders(),
@@ -99,12 +95,7 @@ private static Flux<ByteBuffer> createResponseFlux(StreamResponse sourceResponse
      */
     public Mono<Void> writeValueToAsync(AsynchronousByteChannel channel, ProgressReporter progressReporter) {
         Objects.requireNonNull(channel, "'channel' must not be null");
-        LOGGER.atVerbose()
-            .addKeyValue("thread", Thread.currentThread().getName())
-            .log("BlobDownloadAsyncResponse.writeValueToAsync entry");
         if (sourceResponse != null) {
-            LOGGER.atVerbose()
-                .log("BlobDownloadAsyncResponse.writeValueToAsync using sourceResponse (IOUtils.transfer)");
             return IOUtils.transferStreamResponseToAsynchronousByteChannel(channel, sourceResponse, onErrorResume,
                 progressReporter, retryOptions.getMaxRetryRequests());
         } else if (super.getValue() != null) {

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

@@ -1283,6 +1283,10 @@ Mono<BlobDownloadAsyncResponse> downloadStreamWithResponseInternal(BlobRange ran
                 BlobDownloadHeaders blobDownloadHeaders = ModelHelper.populateBlobDownloadHeaders(blobsDownloadHeaders,
                     ModelHelper.getErrorCode(response.getHeaders()));
 
+                /*
+                 * If the customer did not specify a count, they are reading to the end of the blob. Extract this value
+                 * from the response for better book-keeping towards the end.
+                 */
                 long finalCount;
                 long initialOffset = finalRange.getOffset();
                 if (finalRange.getCount() == null) {
@@ -1292,6 +1296,8 @@ Mono<BlobDownloadAsyncResponse> downloadStreamWithResponseInternal(BlobRange ran
                     finalCount = finalRange.getCount();
                 }
 
+                // The resume function takes throwable and offset at the destination.
+                // I.e. offset is relative to the starting point.
                 BiFunction<Throwable, Long, Mono<StreamResponse>> onDownloadErrorResume = (throwable, offset) -> {
                     if (!(throwable instanceof IOException || throwable instanceof TimeoutException)) {
                         return Mono.error(throwable);
@@ -1539,6 +1545,7 @@ Mono<Response<BlobProperties>> downloadToFileWithResponse(BlobDownloadToFileOpti
         BlobRequestConditions finalConditions
             = options.getRequestConditions() == null ? new BlobRequestConditions() : options.getRequestConditions();
 
+        // Default behavior is not to overwrite
         Set<OpenOption> openOptions = options.getOpenOptions();
         if (openOptions == null) {
             openOptions = DEFAULT_OPEN_OPTIONS_SET;

sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobMessageAsyncDecoderDownloadTests.java

@@ -252,15 +252,13 @@ public void interruptAndVerifyProperRewind() throws IOException {
 
         DownloadRetryOptions retryOptions = new DownloadRetryOptions().setMaxRetryRequests(5);
 
-        StepVerifier.create(downloadClient
-            .downloadStreamWithResponse(new BlobDownloadStreamOptions().setDownloadRetryOptions(retryOptions)
-                .setResponseChecksumAlgorithm(StorageChecksumAlgorithm.CRC64))
-            .doFinally(signalType -> {
-                System.out.println("[MockPartialResponsePolicy] hits=" + mockPolicy.getHits() + ", triesRemaining="
-                    + mockPolicy.getTriesRemaining() + ", ranges=" + mockPolicy.getRangeHeaders());
-                assertTrue(mockPolicy.getHits() > 0, "Mock interruption policy was not invoked");
-            })
-            .flatMap(r -> FluxUtil.collectBytesInByteBufferStream(r.getValue())))
+        StepVerifier
+            .create(downloadClient
+                .downloadStreamWithResponse(new BlobDownloadStreamOptions().setDownloadRetryOptions(retryOptions)
+                    .setResponseChecksumAlgorithm(StorageChecksumAlgorithm.CRC64))
+                .doFinally(
+                    signalType -> assertTrue(mockPolicy.getHits() > 0, "Mock interruption policy was not invoked"))
+                .flatMap(r -> FluxUtil.collectBytesInByteBufferStream(r.getValue())))
             .assertNext(result -> TestUtils.assertArraysEqual(randomData, result))
             .verifyComplete();
 

sdk/storage/azure-storage-common/src/test-shared/java/com/azure/storage/common/test/shared/policy/MockPartialResponsePolicy.java

@@ -94,8 +94,6 @@ public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineN
                     return Mono.just(response);
                 }
                 hits.incrementAndGet();
-                System.out.println("[MockPartialResponsePolicy] invoked. tries=" + remainingTries
-                    + ", maxBytesPerResponse=" + maxBytesPerResponse);
 
                 Flux<ByteBuffer> limitedBody = limitStreamToBytes(response.getBody(), maxBytesPerResponse);
                 return Mono.just(

sdk/storage/azure-storage-common/src/test/java/com/azure/storage/common/implementation/contentvalidation/StructuredMessageDecoderTests.java

@@ -3,6 +3,7 @@
 
 package com.azure.storage.common.implementation.contentvalidation;
 
+import com.azure.core.util.FluxUtil;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Flux;
 
@@ -25,13 +26,7 @@
 public class StructuredMessageDecoderTests {
 
     private static ByteBuffer collectFlux(Flux<ByteBuffer> flux) {
-        ByteArrayOutputStream out = new ByteArrayOutputStream();
-        flux.toIterable().forEach(buf -> {
-            byte[] bytes = new byte[buf.remaining()];
-            buf.get(bytes);
-            out.write(bytes, 0, bytes.length);
-        });
-        return ByteBuffer.wrap(out.toByteArray()).order(ByteOrder.LITTLE_ENDIAN);
+        return ByteBuffer.wrap(FluxUtil.collectBytesInByteBufferStream(flux).block()).order(ByteOrder.LITTLE_ENDIAN);
     }
 
     @Test
@@ -127,9 +122,6 @@ public void readsSegmentHeaderSplitAcrossChunks() throws IOException {
     public void handlesZeroLengthSegment() throws IOException {
         // Test: Zero-length segment should decode correctly
         // Note: Zero-length segments are valid in the format
-        byte[] originalData = new byte[0];
-
-        // For zero-length data, encoder behavior varies - let's test with minimal data
         byte[] minimalData = new byte[1];
         ThreadLocalRandom.current().nextBytes(minimalData);