Fix checkstyle errors and add advanceMessageOffset method

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

@@ -68,6 +68,23 @@ public long getMessageOffset() {
         return messageOffset;
     }
 
+    /**
+     * Advances the message offset by the specified number of bytes.
+     * This should be called after consuming an encoded segment to maintain
+     * the authoritative encoded offset.
+     *
+     * @param bytes The number of bytes to advance.
+     */
+    public void advanceMessageOffset(long bytes) {
+        long priorOffset = messageOffset;
+        messageOffset += bytes;
+        LOGGER.atInfo()
+            .addKeyValue("priorOffset", priorOffset)
+            .addKeyValue("bytesAdvanced", bytes)
+            .addKeyValue("newOffset", messageOffset)
+            .log("Advanced message offset");
+    }
+
     /**
      * Resets the decoder position to the last complete segment boundary.
      * This is used during smart retry to ensure the decoder is in sync with
@@ -139,8 +156,9 @@ private static String toHex(ByteBuffer buf, int len) {
         StringBuilder sb = new StringBuilder();
         for (int i = 0; i < out.length; i++) {
             sb.append(String.format("%02X", out[i]));
-            if (i < out.length - 1)
+            if (i < out.length - 1) {
                 sb.append(' ');
+            }
         }
         return sb.toString();
     }
@@ -175,11 +193,11 @@ public StructuredMessageFlags getFlags() {
      * 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) {
+        final int segmentSizeBytes = 8; // segment size is 8 bytes (long)
+        if (buffer.remaining() < segmentSizeBytes) {
             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");
+            throw LOGGER.logExceptionAsError(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
@@ -200,8 +218,8 @@ private long readAndValidateSegmentLength(ByteBuffer buffer, long remaining) {
                 "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 + ")");
+            throw LOGGER.logExceptionAsError(new IllegalArgumentException("Invalid segment size detected: "
+                + segmentLength + " (remaining=" + remaining + ", decoderOffset=" + messageOffset + ")"));
         }
 
         LOGGER.atVerbose()
@@ -394,14 +412,18 @@ public ByteBuffer decode(ByteBuffer buffer) {
     }
 
     /**
-     * Finalizes the decoding process and validates that the entire message has been decoded.
+     * Finalizes the decoding process and returns any final decoded bytes still buffered internally.
+     * The policy should aggregate decoded byte counts and perform the final length comparison.
      *
-     * @throws IllegalArgumentException if the decoded message length does not match the expected length.
+     * @return A ByteBuffer containing any final decoded bytes, or null if none remain.
+     * @throws IllegalArgumentException if the encoded message offset doesn't match expected length.
      */
-    public void finalizeDecoding() {
+    public ByteBuffer finalizeDecoding() {
         if (messageOffset != messageLength) {
             throw LOGGER.logExceptionAsError(new IllegalArgumentException("Decoded message length does not match "
                 + "expected length. Expected: " + messageLength + ", but was: " + messageOffset));
         }
+        // No buffered decoded bytes in current implementation
+        return null;
     }
 }

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

@@ -125,7 +125,8 @@ private Flux<ByteBuffer> decodeStream(Flux<ByteBuffer> encodedFlux, DecoderState
                         LOGGER.error(
                             "Negative relative index detected: relativeIndex={}, decoderOffset={}, absoluteStart={}",
                             relativeIndex, decoderOffset, absoluteStartOfCombined);
-                        throw new IllegalStateException("Negative relative index: " + relativeIndex);
+                        throw LOGGER.logExceptionAsError(
+                            new IllegalStateException("Negative relative index: " + relativeIndex));
                     }
 
                     // Check if we have enough for segment header
@@ -186,21 +187,33 @@ private Flux<ByteBuffer> decodeStream(Flux<ByteBuffer> encodedFlux, DecoderState
                     // Decode the segment
                     ByteBuffer decoded = state.decoder.decode(encodedSlice);
 
-                    LOGGER.atVerbose()
+                    // Track decoded bytes - update counter regardless of whether bytes were produced
+                    int decodedProduced = (decoded != null) ? decoded.remaining() : 0;
+                    
+                    LOGGER.atInfo()
                         .addKeyValue("relativeIndex", relativeIndex)
                         .addKeyValue("encodedSegmentSize", encodedSegmentSize)
-                        .addKeyValue("decodedBytes", decoded.remaining())
-                        .addKeyValue("newDecoderOffset", state.decoder.getMessageOffset())
+                        .addKeyValue("decodedProduced", decodedProduced)
+                        .addKeyValue("totalDecodedSoFar", state.totalBytesDecoded.get())
+                        .addKeyValue("decoderOffset", state.decoder.getMessageOffset())
                         .log("Decoded segment");
 
                     // Update tracked bytes
                     state.totalEncodedBytesProcessed.addAndGet(encodedSegmentSize);
-                    if (decoded.remaining() > 0) {
-                        state.totalBytesDecoded.addAndGet(decoded.remaining());
+                    
+                    // Always update decoded byte counter (even if zero bytes produced)
+                    if (decodedProduced > 0) {
+                        state.totalBytesDecoded.addAndGet(decodedProduced);
                         // Accumulate decoded bytes
-                        byte[] decodedBytes = new byte[decoded.remaining()];
+                        byte[] decodedBytes = new byte[decodedProduced];
                         decoded.get(decodedBytes);
                         decodedOutput.write(decodedBytes, 0, decodedBytes.length);
+                    } else {
+                        // Log when no decoded bytes are produced from a segment
+                        LOGGER.atInfo()
+                            .addKeyValue("encodedSegmentSize", encodedSegmentSize)
+                            .addKeyValue("decoderOffset", state.decoder.getMessageOffset())
+                            .log("No decoded bytes produced from segment");
                     }
 
                     // Check if we've completed the message