core: Keep getByteBuffer() support when coalescing small buffers

core/src/main/java/io/grpc/internal/CompositeReadableBuffer.java

@@ -123,8 +123,9 @@ private void coalesceTailSmallBuffers(int count) {
       b.close();
     }
 
-    // Wrap and enqueue the single compacted buffer back at the tail
-    ReadableBuffer singleBuffer = ReadableBuffers.wrap(coalescedBytes);
+    // Wrap and enqueue the single compacted buffer back at the tail. Wrap it as a ByteBuffer,
+    // since wrap(byte[]) doesn't support getByteBuffer() and would disable it for the composite.
+    ReadableBuffer singleBuffer = ReadableBuffers.wrap(ByteBuffer.wrap(coalescedBytes));
     readableBuffers.add(singleBuffer);
 
     // Note: The global `readableBytes` remains perfectly synced since we

core/src/test/java/io/grpc/internal/CompositeReadableBufferTest.java

@@ -28,6 +28,7 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.nio.InvalidMarkException;
 import org.junit.After;
 import org.junit.Before;
@@ -336,6 +337,33 @@ public void coalesceBeforeLargeBuffer() {
     assertEquals(1524, composite.readableBytes());
   }
 
+  @Test
+  public void coalesceBeforeLargeBufferKeepsByteBufferSupport() {
+    composite = new CompositeReadableBuffer();
+    // Small frames that support getByteBuffer(), as Netty's do
+    composite.addBuffer(ReadableBuffers.wrap(ByteBuffer.wrap("hello ".getBytes(UTF_8))));
+    composite.addBuffer(ReadableBuffers.wrap(ByteBuffer.wrap("world".getBytes(UTF_8))));
+    composite.addBuffer(ReadableBuffers.wrap(ByteBuffer.wrap(new byte[1024])));
+    assertEquals(2, composite.getBufferCount());
+
+    assertTrue(composite.byteBufferSupported());
+    assertEquals(ByteBuffer.wrap("hello world".getBytes(UTF_8)), composite.getByteBuffer());
+  }
+
+  @Test
+  public void coalesceOnMaxSmallBuffersKeepsByteBufferSupport() {
+    composite = new CompositeReadableBuffer();
+    byte[] expected = new byte[1000];
+    for (int i = 0; i < 1000; i++) {
+      expected[i] = (byte) i;
+      composite.addBuffer(ReadableBuffers.wrap(ByteBuffer.wrap(new byte[] {(byte) i})));
+    }
+    assertEquals(1, composite.getBufferCount());
+
+    assertTrue(composite.byteBufferSupported());
+    assertEquals(ByteBuffer.wrap(expected), composite.getByteBuffer());
+  }
+
   @Test
   public void largeBufferResetsTailSmallBufferCount() {
     composite = new CompositeReadableBuffer();