Remove duplicate byte skipping logic - use simpler approach

Reverted to simpler implementation that doesn't try to skip duplicate bytes. The duplicate-skipping logic was causing "Unexpected segment number" errors because it was interfering with the decoder's state management.

The correct approach:
1. BlobAsyncClientBase requests retry from correct encoded offset (via totalEncodedBytesProcessed)
2. Server sends bytes starting from that offset (no duplicates)
3. Policy feeds all received bytes to decoder with pending buffer handling
4. Decoder maintains state and processes bytes sequentially

Removed:
- getResponseStartOffset() method
- currentResponseStartOffset and currentResponseBytesRead fields from DecoderState
- Duplicate byte detection and skipping logic in decodeStream()

This simpler approach relies on correct offset calculation in the retry logic (which we have) rather than trying to detect and skip duplicates at the policy level.

Addresses comment #2499104452 - investigating test failures.

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

Author: Copilot

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

@@ -63,10 +63,6 @@ public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineN
                 // Get or create decoder with state tracking
                 DecoderState decoderState = getOrCreateDecoderState(context, contentLength);
 
-                // Determine the starting offset of this response (for retry handling)
-                long responseStartOffset = getResponseStartOffset(httpResponse);
-                decoderState.setCurrentResponseStartOffset(responseStartOffset);
-
                 // Decode using the stateful decoder
                 Flux<ByteBuffer> decodedStream = decodeStream(httpResponse.getBody(), decoderState);
 
@@ -90,42 +86,15 @@ public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineN
     private Flux<ByteBuffer> decodeStream(Flux<ByteBuffer> encodedFlux, DecoderState state) {
         return encodedFlux.concatMap(encodedBuffer -> {
             try {
-                // Calculate the absolute offset of this buffer in the encoded stream
-                long bufferAbsoluteOffset = state.currentResponseStartOffset + state.currentResponseBytesRead;
-                int bufferSize = encodedBuffer.remaining();
-                state.currentResponseBytesRead += bufferSize;
-
-                // Determine how many bytes to skip (if this is overlapping with already-processed data)
-                long bytesToSkip = state.totalEncodedBytesProcessed.get() - bufferAbsoluteOffset;
-
-                ByteBuffer bufferToProcess;
-                if (bytesToSkip > 0) {
-                    // Skip bytes that have already been processed
-                    if (bytesToSkip >= bufferSize) {
-                        // Entire buffer is duplicate, skip it
-                        return Flux.empty();
-                    }
-                    // Skip partial buffer
-                    ByteBuffer skippedBuffer = encodedBuffer.duplicate();
-                    skippedBuffer.position(skippedBuffer.position() + (int) bytesToSkip);
-                    bufferToProcess = skippedBuffer;
-                    // Only count the new bytes
-                    state.totalEncodedBytesProcessed.addAndGet(bufferSize - (int) bytesToSkip);
-                } else {
-                    // All bytes are new
-                    bufferToProcess = encodedBuffer;
-                    state.totalEncodedBytesProcessed.addAndGet(bufferSize);
-                }
-
                 // Combine with pending data if any
-                ByteBuffer dataToProcess = state.combineWithPending(bufferToProcess);
+                ByteBuffer dataToProcess = state.combineWithPending(encodedBuffer);
+
+                // Track encoded bytes
+                int encodedBytesInBuffer = encodedBuffer.remaining();
+                state.totalEncodedBytesProcessed.addAndGet(encodedBytesInBuffer);
 
                 // Try to decode what we have - decoder handles partial data
                 int availableSize = dataToProcess.remaining();
-                if (availableSize == 0) {
-                    return Flux.empty();
-                }
-
                 ByteBuffer decodedData = state.decoder.decode(dataToProcess.duplicate(), availableSize);
 
                 // Track decoded bytes
@@ -217,30 +186,6 @@ private DecoderState getOrCreateDecoderState(HttpPipelineCallContext context, lo
             .orElseGet(() -> new DecoderState(contentLength));
     }
 
-    /**
-     * Gets the starting offset of the response by parsing the Content-Range header.
-     *
-     * @param httpResponse The HTTP response.
-     * @return The starting offset, or 0 if not a range response.
-     */
-    private long getResponseStartOffset(HttpResponse httpResponse) {
-        String contentRange = httpResponse.getHeaderValue(HttpHeaderName.CONTENT_RANGE);
-        if (contentRange != null && contentRange.startsWith("bytes ")) {
-            try {
-                // Format: "bytes start-end/total" or "bytes start-end/*"
-                String range = contentRange.substring(6); // Skip "bytes "
-                int dashIndex = range.indexOf('-');
-                if (dashIndex > 0) {
-                    String startStr = range.substring(0, dashIndex);
-                    return Long.parseLong(startStr);
-                }
-            } catch (Exception e) {
-                LOGGER.warning("Failed to parse Content-Range header: " + contentRange, e);
-            }
-        }
-        return 0;
-    }
-
     /**
      * Checks if the response is a download response.
      *
@@ -262,8 +207,6 @@ public static class DecoderState {
         private final AtomicLong totalBytesDecoded;
         private final AtomicLong totalEncodedBytesProcessed;
         private ByteBuffer pendingBuffer;
-        private long currentResponseStartOffset = 0;
-        private long currentResponseBytesRead = 0;
 
         /**
          * Creates a new decoder state.
@@ -278,16 +221,6 @@ public DecoderState(long expectedContentLength) {
             this.pendingBuffer = null;
         }
 
-        /**
-         * Sets the starting offset of the current response (for retry handling).
-         *
-         * @param offset The starting offset.
-         */
-        public void setCurrentResponseStartOffset(long offset) {
-            this.currentResponseStartOffset = offset;
-            this.currentResponseBytesRead = 0;
-        }
-
         /**
          * Combines pending buffer with new data.
          *