core: Support getByteBuffer() on byte array ReadableBuffers

[merlimat]

Problem

Since grpc/grpc-java#12924 (also in 1.82.4, 1.83.1 and 1.84.x), CompositeReadableBuffer.enqueueBuffer() merges runs of small (< 1 KiB) buffers. A merge happens when 1000 have queued, or when a buffer of at least 1 KiB follows two or more of them. coalesceTailSmallBuffers() wraps the merged bytes with ReadableBuffers.wrap(byte[]). The resulting ByteArrayWrapper inherits byteBufferSupported() == false from AbstractReadableBuffer. CompositeReadableBuffer.byteBufferSupported() requires every component to support it, so after a merge the message stream from MessageDeframer reports no HasByteBuffer access, even though Netty and Cronet frames support it.

This affects custom marshallers that parse in place through HasByteBuffer/Detachable (grpc/grpc-java#8102). They silently fall back to copying whenever a message picks up merged small frames. For example, in LightProto's GrpcNettyTransportTest (streamnative/lightproto#48), a ~64 KiB response that starts with merged small buffers is copied instead of parsed in place on every run.

Example

CompositeReadableBuffer composite = new CompositeReadableBuffer();
// Two small frames that support getByteBuffer(), then a large one, as Netty may deliver them
composite.addBuffer(ReadableBuffers.wrap(ByteBuffer.wrap(new byte[100])));
composite.addBuffer(ReadableBuffers.wrap(ByteBuffer.wrap(new byte[100])));
composite.addBuffer(ReadableBuffers.wrap(ByteBuffer.wrap(new byte[2048])));

composite.byteBufferSupported(); // false before this change, true after

The large buffer triggers the merge:

// CompositeReadableBuffer.coalesceTailSmallBuffers()
ReadableBuffer singleBuffer = ReadableBuffers.wrap(coalescedBytes); // ByteArrayWrapper

// CompositeReadableBuffer.byteBufferSupported()
for (ReadableBuffer buffer : readableBuffers) {
  if (!buffer.byteBufferSupported()) { // ByteArrayWrapper: false
    return false;
  }
}

Change

ReadableBuffers.ByteArrayWrapper now supports getByteBuffer():

@Override
public boolean byteBufferSupported() {
  return true;
}

@Override
public ByteBuffer getByteBuffer() {
  return ByteBuffer.wrap(bytes, offset, end - offset).slice();
}

This is the same expression Netty's UnpooledHeapByteBuf.nioBuffer() uses. It follows the ReadableBuffer.getByteBuffer() contract the same way ByteReadableBufferWrapper (bytes.slice()) does:

• No copy: the view shares the backing array. Like the other implementations, it is not read-only. A read-only heap ByteBuffer reports hasArray() == false, so consumers such as protobuf's CodedInputStream.newInstance(ByteBuffer) would copy it.
• It doesn't move this buffer's read position, and it has its own position, limit and mark. slice() limits it to the unread bytes, so clear() or rewind() can't expose bytes already read, or bytes owned by another wrapper on the same array (readBytes(n) slices).
• mark()/reset() only move offset, so getByteBuffer() after reset() starts at the mark again.
• An exhausted wrapper returns an empty buffer, like the other leaf buffers, and CompositeReadableBuffer still returns null once drained. A leaf returning null would end a composite's content early if an empty buffer were at its head.

Every array wrapped this way is never written again after wrapping. That holds for merged buffers, the servlet transport's copied chunks, and the MessageDeframer inflate buffer regions, so a marshaller that detaches and keeps the views is safe.

No in-repo code consumes HasByteBuffer, so beyond the fix, only custom marshallers see these changes:

• Servlet transport (javax and jakarta): inbound chunks are ReadableBuffers.wrap(Arrays.copyOf(...)), so servlet messages now report ByteBuffer access. Before, they never did.
• Full-stream decompression: messages built from ReadableBuffers.wrap(inflatedBuffer, inflatedIndex, n) now report it too. This path has had no public switch since grpc/grpc-java#10744.
• OkHttp: OkHttpReadableBuffer still doesn't support it. Only a message made up entirely of merged chunks now reports true.
• BufferInputStream.detach() leaves ReadableBuffers.empty() behind for composite-backed streams. That stream now returns an empty buffer from getByteBuffer() instead of reporting no support.
• In-process and binder don't use ReadableBuffer, so they're unchanged.

Alternative: fix only the regression by having coalesceTailSmallBuffers() use ReadableBuffers.wrap(ByteBuffer.wrap(coalescedBytes)), a ByteReadableBufferWrapper. That keeps the servlet and decompression paths as they are, but leaves wrap(byte[]) as the only leaf without ByteBuffer access. Happy to switch if that's preferred.

Testing

• ReadableBuffersArrayTest: new tests for a no-copy view with an array offset, the view after partial reads and skipBytes() (empty once exhausted), readBytes(n) slices, and mark()/reset(). The inherited ReadableBufferTestBase ByteBuffer tests now run for the array wrapper instead of being skipped.
• CompositeReadableBufferTest: two small buffers before a 1 KiB buffer, and 1000 small buffers. Both check that byteBufferSupported() stays true and that getByteBuffer() returns the merged bytes in order.
• The six new tests fail without the fix. With it, ./gradlew -PskipCodegen=true -PskipAndroid=true :grpc-core:check passes (all grpc-core tests, checkstyle, animal-sniffer), as do :grpc-netty:test, :grpc-okhttp:test and :grpc-servlet:test, on JDK 17.