netty: Touch ByteBuf when message framing has been decoded

When a memory leak occurs, it is really helpful to have access records
to understand where the buffer was being held when it leaked. retain()
when we create the NettyReadableBuffer already creates an access record
the ByteBuf, so here we track when the ByteBuf is passed to another
thread.

See #8330

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

@@ -97,6 +97,11 @@ public abstract class ForwardingReadableBuffer implements ReadableBuffer {
     return buf.arrayOffset();
   }
 
+  @Override
+  public void touch() {
+    buf.touch();
+  }
+
   @Override
   public boolean markSupported() {
     return buf.markSupported();

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

@@ -409,6 +409,7 @@ public class MessageDeframer implements Closeable, Deframer {
     statsTraceCtx.inboundMessageRead(currentMessageSeqNo, inboundBodyWireSize, -1);
     inboundBodyWireSize = 0;
     InputStream stream = compressedFlag ? getCompressedBody() : getUncompressedBody();
+    nextFrame.touch();
     nextFrame = null;
     listener.messagesAvailable(new SingleMessageProducer(stream));
 

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

@@ -124,6 +124,14 @@ public interface ReadableBuffer extends Closeable {
    */
   int arrayOffset();
 
+  /**
+   * Note that the current callsite has access to this buffer, or do nothing. This is only useful
+   * when the buffer has leak detection and intrumentation to record usages before the buffer was
+   * leaked. That can make it much easier to track down where the buffer was leaked. If this isn't
+   * such a buffer, the method does nothing.
+   */
+  default void touch() {}
+
   /**
    * Indicates whether or not {@link #mark} operation is supported for this buffer.
    */

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

@@ -94,6 +94,11 @@ class NettyReadableBuffer extends AbstractReadableBuffer {
     return buffer.arrayOffset() + buffer.readerIndex();
   }
 
+  @Override
+  public void touch() {
+    buffer.touch();
+  }
+
   @Override
   public boolean markSupported() {
     return true;