Add comprehensive diagnostic logging for smart retry debugging

Added detailed logging to track decoder state and buffer positions:
- readAndValidateSegmentLength() with hex dump of buffer contents
- Enhanced resetToLastCompleteSegment() logging with all state details
- Segment completion logging with offset and length details
- Buffer reception logging in decodeStream with pending/new byte counts
- Retry offset calculation logging with before/after decoder state

This will help identify where decoder state and retry offset become misaligned.

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

Author: Copilot

sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/implementation/structuredmessage/StructuredMessageDecoder.java

@@ -75,12 +75,21 @@ public long getMessageOffset() {
      */
     public void resetToLastCompleteSegment() {
         if (messageOffset != lastCompleteSegmentStart) {
-            LOGGER.verbose("Resetting decoder from offset {} to last complete segment boundary {}", messageOffset,
-                lastCompleteSegmentStart);
+            LOGGER.atInfo()
+                .addKeyValue("fromOffset", messageOffset)
+                .addKeyValue("toOffset", lastCompleteSegmentStart)
+                .addKeyValue("currentSegmentNum", currentSegmentNumber)
+                .addKeyValue("currentSegmentContentOffset", currentSegmentContentOffset)
+                .addKeyValue("currentSegmentContentLength", currentSegmentContentLength)
+                .log("Resetting decoder to last complete segment boundary");
             messageOffset = lastCompleteSegmentStart;
             // Reset current segment state - next decode will read the segment header
             currentSegmentContentOffset = 0;
             currentSegmentContentLength = 0;
+        } else {
+            LOGGER.atVerbose()
+                .addKeyValue("offset", messageOffset)
+                .log("Decoder already at last complete segment boundary, no reset needed");
         }
     }
 
@@ -118,6 +127,65 @@ private void readMessageHeader(ByteBuffer buffer) {
         messageOffset += V1_HEADER_LENGTH;
     }
 
+    /**
+     * Converts a ByteBuffer range to hex string for diagnostic purposes.
+     */
+    private static String toHex(ByteBuffer buf, int len) {
+        int pos = buf.position();
+        int peek = Math.min(len, buf.remaining());
+        byte[] out = new byte[peek];
+        buf.get(out, 0, peek);
+        buf.position(pos);
+        StringBuilder sb = new StringBuilder();
+        for (int i = 0; i < out.length; i++) {
+            sb.append(String.format("%02X", out[i]));
+            if (i < out.length - 1)
+                sb.append(' ');
+        }
+        return sb.toString();
+    }
+
+    /**
+     * Reads and validates segment length with diagnostic logging.
+     */
+    private long readAndValidateSegmentLength(ByteBuffer buffer, long remaining) {
+        final int SEGMENT_SIZE_BYTES = 8; // segment size is 8 bytes (long)
+        if (buffer.remaining() < SEGMENT_SIZE_BYTES) {
+            LOGGER.error("Not enough bytes to read segment size. pos={}, remaining={}", buffer.position(),
+                buffer.remaining());
+            throw new IllegalStateException("Not enough bytes to read segment size");
+        }
+
+        // Diagnostic: dump first 16 bytes at this position so we can see what's being read
+        LOGGER.atInfo()
+            .addKeyValue("decoderOffset", messageOffset)
+            .addKeyValue("bufferPos", buffer.position())
+            .addKeyValue("bufferRemaining", buffer.remaining())
+            .addKeyValue("peek16", toHex(buffer, 16))
+            .addKeyValue("lastCompleteSegment", lastCompleteSegmentStart)
+            .log("Decoder about to read segment length");
+
+        long segmentLength = buffer.getLong();
+
+        if (segmentLength < 0 || segmentLength > remaining) {
+            // Peek next bytes for extra detail
+            String peekNext = toHex(buffer, 16);
+            LOGGER.error(
+                "Invalid segment length read: segmentLength={}, remaining={}, decoderOffset={}, "
+                    + "lastCompleteSegment={}, bufferPos={}, peek-next-bytes={}",
+                segmentLength, remaining, messageOffset, lastCompleteSegmentStart, buffer.position(), peekNext);
+            throw new IllegalArgumentException("Invalid segment size detected: " + segmentLength + " (remaining="
+                + remaining + ", decoderOffset=" + messageOffset + ")");
+        }
+
+        LOGGER.atVerbose()
+            .addKeyValue("segmentLength", segmentLength)
+            .addKeyValue("decoderOffset", messageOffset)
+            .log("Valid segment length read");
+
+        return segmentLength;
+    }
+
     /**
      * Reads the segment header from the given buffer.
      *
@@ -134,14 +202,8 @@ private void readSegmentHeader(ByteBuffer buffer) {
 
         int segmentNum = Short.toUnsignedInt(buffer.getShort());
 
-        // Read segment size as long (8 bytes)
-        long segmentSize = buffer.getLong();
-
-        // Validate segment size
-        if (segmentSize < 0L || segmentSize > buffer.remaining()) {
-            throw LOGGER
-                .logExceptionAsError(new IllegalArgumentException("Invalid segment size detected: " + segmentSize));
-        }
+        // Read segment size with validation and diagnostics
+        long segmentSize = readAndValidateSegmentLength(buffer, buffer.remaining());
 
         if (segmentNum != currentSegmentNumber + 1) {
             throw LOGGER.logExceptionAsError(new IllegalArgumentException("Unexpected segment number."));
@@ -230,7 +292,11 @@ private void readSegmentFooter(ByteBuffer buffer) {
         // Mark that this segment is complete - update the last complete segment boundary
         // This is the position where we can safely resume if a retry occurs
         lastCompleteSegmentStart = messageOffset;
-        LOGGER.verbose("Segment {} complete at byte offset {}", currentSegmentNumber, lastCompleteSegmentStart);
+        LOGGER.atInfo()
+            .addKeyValue("segmentNum", currentSegmentNumber)
+            .addKeyValue("offset", lastCompleteSegmentStart)
+            .addKeyValue("segmentLength", currentSegmentContentLength)
+            .log("Segment complete at byte offset");
 
         if (currentSegmentNumber == numSegments) {
             readMessageFooter(buffer);

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

@@ -89,6 +89,15 @@ private Flux<ByteBuffer> decodeStream(Flux<ByteBuffer> encodedFlux, DecoderState
             int newBytesReceived = encodedBuffer.remaining();
             state.totalEncodedBytesProcessed.addAndGet(newBytesReceived);
 
+            int pendingSize = (state.pendingBuffer != null) ? state.pendingBuffer.remaining() : 0;
+            LOGGER.atInfo()
+                .addKeyValue("newBytes", newBytesReceived)
+                .addKeyValue("pendingBytes", pendingSize)
+                .addKeyValue("totalProcessed", state.totalEncodedBytesProcessed.get())
+                .addKeyValue("decoderOffset", state.decoder.getMessageOffset())
+                .addKeyValue("lastCompleteSegment", state.decoder.getLastCompleteSegmentStart())
+                .log("Received buffer in decodeStream");
+
             // Combine with pending data if any
             ByteBuffer dataToProcess = state.combineWithPending(encodedBuffer);
 
@@ -309,6 +318,15 @@ public long getRetryOffset() {
             // Use the decoder's last complete segment start as the retry offset
             // This ensures we resume from a segment boundary, not mid-segment
             long retryOffset = decoder.getLastCompleteSegmentStart();
+            long decoderOffsetBefore = decoder.getMessageOffset();
+            int pendingSize = (pendingBuffer != null) ? pendingBuffer.remaining() : 0;
+
+            LOGGER.atInfo()
+                .addKeyValue("retryOffset", retryOffset)
+                .addKeyValue("decoderOffsetBefore", decoderOffsetBefore)
+                .addKeyValue("pendingBytes", pendingSize)
+                .addKeyValue("totalProcessed", totalEncodedBytesProcessed.get())
+                .log("Computing retry offset");
 
             // Reset decoder to the last complete segment boundary
             // This ensures messageOffset and segment state match the retry offset
@@ -317,12 +335,16 @@ public long getRetryOffset() {
             // Clear pending buffer since we're restarting from the segment boundary
             // Any bytes in pending are from after this boundary and will be re-fetched
             if (pendingBuffer != null && pendingBuffer.hasRemaining()) {
-                LOGGER.verbose("Clearing {} pending bytes for retry from segment boundary {}",
-                    pendingBuffer.remaining(), retryOffset);
+                LOGGER.atInfo()
+                    .addKeyValue("pendingBytes", pendingBuffer.remaining())
+                    .addKeyValue("retryOffset", retryOffset)
+                    .log("Clearing pending bytes for retry from segment boundary");
                 pendingBuffer = null;
             }
 
-            LOGGER.verbose("Retry offset calculated: {} (last complete segment boundary)", retryOffset);
+            LOGGER.atInfo()
+                .addKeyValue("retryOffset", retryOffset)
+                .log("Retry offset calculated (last complete segment boundary)");
             return retryOffset;
         }