Release transport messages incrementally while reading them

Still a few small steps left to clean this up but even in this
form this solution essentially up to halves the heap used for
handling large bulk shard requests on non-coordinating data nodes (this is
just one example, there's a couple spots where this saves a lot of memory).
Also, this could be extended to be a little smarter around compression easily, allowing
for potential order of magnitude savings around indexing if we lazy deserialize
individual docs or play similar tricks.

Author: original-brownbear

server/src/main/java/org/elasticsearch/common/bytes/ReleasableBytesReference.java

@@ -259,30 +259,14 @@ public void close() {
                 }
             }
 
-            @Override
-            public int read(byte[] b, int bOffset, int len) throws IOException {
-                int res = super.read(b, bOffset, len);
+            public void tryDiscard() {
                 if (markEnd == 0) {
-                    tryDiscard();
-                }
-                return res;
-            }
-
-            private void tryDiscard() {
-                if (bytesReference instanceof CompositeBytesReference c) {
-                    maybeDiscardReadBytes(c.components());
-                } else if (available() == 0) {
-                    close();
-                }
-            }
-
-            @Override
-            public int read() throws IOException {
-                int res = super.read();
-                if (res == -1 && markEnd == 0) {
-                    close();
+                    if (bytesReference instanceof CompositeBytesReference c) {
+                        maybeDiscardReadBytes(c.components());
+                    } else if (available() == 0) {
+                        close();
+                    }
                 }
-                return res;
             }
 
             private int markEnd = 0;

server/src/main/java/org/elasticsearch/common/io/stream/FilterStreamInput.java

@@ -108,6 +108,11 @@ public int read(byte[] b, int off, int len) throws IOException {
         return delegate.read(b, off, len);
     }
 
+    @Override
+    public void tryDiscard() {
+        delegate.tryDiscard();
+    }
+
     @Override
     public void close() throws IOException {
         delegate.close();

server/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java

@@ -109,13 +109,17 @@ public void setTransportVersion(TransportVersion version) {
     @Override
     public abstract int read(byte[] b, int off, int len) throws IOException;
 
+    public void tryDiscard() {}
+
     /**
      * Reads a bytes reference from this stream, copying any bytes read to a new {@code byte[]}. Use {@link #readReleasableBytesReference()}
      * when reading large bytes references where possible top avoid needless allocations and copying.
      */
     public BytesReference readBytesReference() throws IOException {
         int length = readArraySize();
-        return readBytesReference(length);
+        var res = readBytesReference(length);
+        tryDiscard();
+        return res;
     }
 
     /**

server/src/main/java/org/elasticsearch/transport/TransportLogger.java

@@ -118,35 +118,24 @@ private static String format(TcpChannel channel, InboundMessage message, String
         if (message.isPing()) {
             sb.append(" [ping]").append(' ').append(event).append(": ").append(6).append('B');
         } else {
-            boolean success = false;
             Header header = message.getHeader();
             int networkMessageSize = header.getNetworkMessageSize();
             int messageLengthWithHeader = HEADER_SIZE + networkMessageSize;
-            StreamInput streamInput = message.openOrGetStreamInput();
-            try {
-                final long requestId = header.getRequestId();
-                final boolean isRequest = header.isRequest();
-                final String type = isRequest ? "request" : "response";
-                final String version = header.getVersion().toString();
-                sb.append(" [length: ").append(messageLengthWithHeader);
-                sb.append(", request id: ").append(requestId);
-                sb.append(", type: ").append(type);
-                sb.append(", version: ").append(version);
+            final long requestId = header.getRequestId();
+            final boolean isRequest = header.isRequest();
+            final String type = isRequest ? "request" : "response";
+            final String version = header.getVersion().toString();
+            sb.append(" [length: ").append(messageLengthWithHeader);
+            sb.append(", request id: ").append(requestId);
+            sb.append(", type: ").append(type);
+            sb.append(", version: ").append(version);
 
-                // TODO: Maybe Fix for BWC
-                if (header.needsToReadVariableHeader() == false && isRequest) {
-                    sb.append(", action: ").append(header.getActionName());
-                }
-                sb.append(']');
-                sb.append(' ').append(event).append(": ").append(messageLengthWithHeader).append('B');
-                success = true;
-            } finally {
-                if (success) {
-                    IOUtils.close(streamInput);
-                } else {
-                    IOUtils.closeWhileHandlingException(streamInput);
-                }
+            // TODO: Maybe Fix for BWC
+            if (header.needsToReadVariableHeader() == false && isRequest) {
+                sb.append(", action: ").append(header.getActionName());
             }
+            sb.append(']');
+            sb.append(' ').append(event).append(": ").append(messageLengthWithHeader).append('B');
         }
         return sb.toString();
     }