Test and fix the issue with CompositeByteBuf.component().readerIndex()

netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java

@@ -143,6 +143,12 @@ static void mergeWithCompositeTail(
         // Ideal case: the tail isn't shared, and can be expanded to the required capacity.
         // Take ownership of the tail.
         newTail = tail.retain();
+
+        // TODO(sergiitk): remove when netty issue resolved.
+        // Correct the indexes.
+        ByteBuf sliceDuplicate = composite.internalComponent(tailComponentIndex).duplicate();
+        newTail.setIndex(sliceDuplicate.readerIndex(), sliceDuplicate.writerIndex());
+
         /*
          * The tail is a readable non-composite buffer, so writeBytes() handles everything for us.
          *

netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java

@@ -597,4 +597,57 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex,
       }
     }
   }
+
+  /**
+   * Miscellaneous tests for {@link NettyAdaptiveCumulator#mergeWithCompositeTail} that don't
+   * fit into {@link MergeWithCompositeTailTests}, and require custom-crafted scenarios.
+   */
+  @RunWith(JUnit4.class)
+  public static class MergeWithCompositeTailMiscTests {
+    private final ByteBufAllocator alloc = new PooledByteBufAllocator();
+
+    /**
+     * Test the issue with {@link CompositeByteBuf#component(int)} returning a ByteBuf with
+     * the indexes out-of-sync with {@code CompositeByteBuf.Component} offsets.
+     */
+    @Test
+    public void mergeWithCompositeTail_outOfSyncComposite() {
+      NettyAdaptiveCumulator cumulator = new NettyAdaptiveCumulator(1024);
+
+      // Create underlying buffer spacious enough for the test data.
+      ByteBuf buf = alloc.buffer(32).writeBytes("---01234".getBytes(US_ASCII));
+
+      // Start with a regular cumulation and add the buf as the only component.
+      CompositeByteBuf composite1 = alloc.compositeBuffer(8).addFlattenedComponents(true, buf);
+      // Read composite1 buf to the beginning of the numbers.
+      assertThat(composite1.readCharSequence(3, US_ASCII).toString()).isEqualTo("---");
+
+      // Wrap composite1 into another cumulation. This is similar to
+      // what NettyAdaptiveCumulator.cumulate() does in the case the cumulation has refCnt != 1.
+      CompositeByteBuf composite2 =
+          alloc.compositeBuffer(8).addFlattenedComponents(true, composite1);
+      assertThat(composite2.toString(US_ASCII)).isEqualTo("01234");
+
+      // The previous operation does not adjust the read indexes of the underlying buffers,
+      // only the internal Component offsets. When the cumulator attempts to append the input to
+      // the tail buffer, it extracts it from the cumulation, writes to it, and then adds it back.
+      // Because the readerIndex on the tail buffer is not adjusted during the read operation
+      // on the CompositeByteBuf, adding the tail back results in the discarded bytes of the tail
+      // to be added back to the cumulator as if they were never read.
+      //
+      // If the reader index of the tail is not manually corrected, the resulting
+      // cumulation will contain the discarded part of the tail: "---".
+      // If it's corrected, it will only contain the numbers.
+      CompositeByteBuf cumulation = (CompositeByteBuf) cumulator.cumulate(alloc, composite2,
+          ByteBufUtil.writeAscii(alloc, "56789"));
+      assertThat(cumulation.toString(US_ASCII)).isEqualTo("0123456789");
+
+      // Correctness check: we still have a single component, and this component is still the
+      // original underlying buffer.
+      assertThat(cumulation.numComponents()).isEqualTo(1);
+      // Replace '2' with '*', and '8' with '*'.
+      buf.setByte(5, '*').setByte(11, '$');
+      assertThat(cumulation.toString(US_ASCII)).isEqualTo("01*34567$9");
+    }
+  }
 }