Fix smart retry by preserving partial data in pending buffer on decoder exceptions

The root cause of "Unexpected segment number" errors was that when the decoder threw exceptions due to insufficient data (e.g., "Content not long enough"), the partial data was not saved to the pending buffer. This caused retries to lose accumulated bytes and start over.

Key changes:
1. Moved buffer combination and byte tracking outside try block to execute regardless of decoder success/failure
2. Added specific handling for IllegalArgumentException with "not long enough" message
3. When insufficient data error occurs, save all accumulated data to pending buffer and return empty (don't fail)
4. This allows the stream to continue accumulating bytes across retries until enough data is available for decoding

How it works now:
- Request bytes=0-: Get byte 0, insufficient for header, save to pending, return empty
- Stream error triggers retry
- Request bytes=1-: Get byte 1, combine with pending [byte 0], still insufficient, save [0,1] to pending
- Continue until 13+ bytes accumulated in pending
- Eventually enough bytes available, decoder succeeds and processes the header
- Smart retry resumes from correct encoded offset

This fix enables true smart retry where partial data is preserved across network interruptions.

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

Author: Copilot

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

@@ -85,14 +85,14 @@ public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineN
      */
     private Flux<ByteBuffer> decodeStream(Flux<ByteBuffer> encodedFlux, DecoderState state) {
         return encodedFlux.concatMap(encodedBuffer -> {
-            try {
-                // Combine with pending data if any
-                ByteBuffer dataToProcess = state.combineWithPending(encodedBuffer);
+            // Combine with pending data if any
+            ByteBuffer dataToProcess = state.combineWithPending(encodedBuffer);
 
-                // Track encoded bytes
-                int encodedBytesInBuffer = encodedBuffer.remaining();
-                state.totalEncodedBytesProcessed.addAndGet(encodedBytesInBuffer);
+            // Track encoded bytes
+            int encodedBytesInBuffer = encodedBuffer.remaining();
+            state.totalEncodedBytesProcessed.addAndGet(encodedBytesInBuffer);
 
+            try {
                 // Try to decode what we have - decoder handles partial data
                 int availableSize = dataToProcess.remaining();
                 ByteBuffer decodedData = state.decoder.decode(dataToProcess.duplicate(), availableSize);
@@ -114,6 +114,19 @@ private Flux<ByteBuffer> decodeStream(Flux<ByteBuffer> encodedFlux, DecoderState
                 } else {
                     return Flux.empty();
                 }
+            } catch (IllegalArgumentException e) {
+                // Handle decoder exceptions - check if it's due to incomplete data
+                if (e.getMessage() != null && e.getMessage().contains("not long enough")) {
+                    // Not enough data to decode yet - preserve all data in pending buffer
+                    state.updatePendingBuffer(dataToProcess);
+
+                    // Don't fail - just return empty and wait for more data
+                    return Flux.empty();
+                } else {
+                    // Other errors should propagate
+                    LOGGER.error("Failed to decode structured message chunk: " + e.getMessage(), e);
+                    return Flux.error(e);
+                }
             } catch (Exception e) {
                 LOGGER.error("Failed to decode structured message chunk: " + e.getMessage(), e);
                 return Flux.error(e);