Fix smart retry offset calculation to exclude pending buffer bytes

The root cause of "Unexpected segment number" errors was incorrect retry offset calculation. The pending buffer contains bytes that have already been counted in totalEncodedBytesProcessed but haven't been successfully processed by the decoder yet. When retrying, we were requesting bytes AFTER the pending buffer, causing gaps in the data stream.

Key issue:
- totalEncodedBytesProcessed tracks ALL bytes received (including those in pending)
- Pending buffer holds bytes waiting for more data to complete a structure (header/segment/footer)
- Retry offset was set to totalEncodedBytesProcessed, skipping pending bytes
- This caused decoder to receive segment N when expecting segment N-1

Solution:
1. Added getRetryOffset() method to DecoderState that returns: totalEncodedBytesProcessed - pendingBufferSize
2. Updated BlobAsyncClientBase to use getRetryOffset() instead of getTotalEncodedBytesProcessed()
3. Added import for DecoderState inner class

Example flow:
- Receive bytes 0-4, add to totalEncodedBytesProcessed (=5), insufficient for 13-byte header, store in pending
- IOException occurs
- Retry requests from offset 0 (5 - 5 pending bytes = 0) ✓
- Get byte 5, combine with pending [0-4] = [0-5], still insufficient, store in pending
- totalEncodedBytesProcessed now = 6, pending = 6 bytes
- Retry requests from offset 0 (6 - 6 = 0) ✓
- Continue until enough bytes accumulated

This ensures continuous byte stream to decoder with no gaps or duplicates.

Co-authored-by: gunjansingh-msft <[email protected]>

Author: Copilot

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

@@ -86,6 +86,7 @@
 import com.azure.storage.common.implementation.StorageImplUtils;
 import com.azure.storage.common.DownloadContentValidationOptions;
 import com.azure.storage.common.policy.StorageContentValidationDecoderPolicy;
+import com.azure.storage.common.policy.StorageContentValidationDecoderPolicy.DecoderState;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.SignalType;
@@ -1404,11 +1405,11 @@ Mono<BlobDownloadAsyncResponse> downloadStreamWithResponse(BlobRange range, Down
                                     .orElse(null);
 
                             if (decoderStateObj instanceof StorageContentValidationDecoderPolicy.DecoderState) {
-                                StorageContentValidationDecoderPolicy.DecoderState decoderState
-                                    = (StorageContentValidationDecoderPolicy.DecoderState) decoderStateObj;
+                                DecoderState decoderState = (DecoderState) decoderStateObj;
 
-                                // Use the encoded offset for retry (number of encoded bytes processed)
-                                long encodedOffset = decoderState.getTotalEncodedBytesProcessed();
+                                // Use the retry offset (encoded bytes processed minus pending buffer)
+                                // The pending buffer contains bytes that have been counted but not yet successfully processed
+                                long encodedOffset = decoderState.getRetryOffset();
                                 long remainingCount = finalCount - encodedOffset;
                                 retryRange = new BlobRange(initialOffset + encodedOffset, remainingCount);
 

sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/policy/StorageContentValidationDecoderPolicy.java

@@ -282,6 +282,19 @@ public long getTotalEncodedBytesProcessed() {
             return totalEncodedBytesProcessed.get();
         }
 
+        /**
+         * Gets the offset to use for retry requests.
+         * This is the total encoded bytes processed minus any bytes in the pending buffer,
+         * since pending bytes have already been counted but haven't been successfully processed yet.
+         *
+         * @return The offset for retry requests.
+         */
+        public long getRetryOffset() {
+            long processed = totalEncodedBytesProcessed.get();
+            int pending = (pendingBuffer != null) ? pendingBuffer.remaining() : 0;
+            return processed - pending;
+        }
+
         /**
          * Checks if the decoder has finalized.
          *