Improve MockTcpTransport memory usage (#35402) (#35430)

The MockTcpTransport is not friendly in regards to memory usage. It must
allocate multiple byte arrays for every message. This improves the
memory situation by failing fast if the message is improperly formatted.
Additionally, it uses reusable big arrays for at least half of the
allocated byte arrays.

Author: Tim-Brooks

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

@@ -198,7 +198,7 @@ public abstract class TcpTransport extends AbstractLifecycleComponent implements
     protected final ScheduledPing scheduledPing;
     private final TimeValue pingSchedule;
     protected final ThreadPool threadPool;
-    private final BigArrays bigArrays;
+    protected final BigArrays bigArrays;
     protected final NetworkService networkService;
     protected final Set<ProfileSettings> profileSettings;
 

test/framework/src/main/java/org/elasticsearch/transport/MockTcpTransport.java

@@ -18,14 +18,14 @@
  */
 package org.elasticsearch.transport;
 
-import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.InputStreamStreamInput;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.io.stream.ReleasableBytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Settings;
@@ -35,6 +35,7 @@
 import org.elasticsearch.common.util.CancellableThreads;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.mocksocket.MockServerSocket;
 import org.elasticsearch.mocksocket.MockSocket;
@@ -153,19 +154,20 @@ private void readMessage(MockChannel mockChannel, StreamInput input) throws IOEx
         if (msgSize == -1) {
             socket.getOutputStream().flush();
         } else {
-            BytesStreamOutput output = new BytesStreamOutput();
-            final byte[] buffer = new byte[msgSize];
-            input.readFully(buffer);
-            output.write(minimalHeader);
-            output.writeInt(msgSize);
-            output.write(buffer);
-            final BytesReference bytes = output.bytes();
-            if (TcpTransport.validateMessageHeader(bytes)) {
-                InetSocketAddress remoteAddress = (InetSocketAddress) socket.getRemoteSocketAddress();
-                messageReceived(bytes.slice(TcpHeader.MARKER_BYTES_SIZE + TcpHeader.MESSAGE_LENGTH_SIZE, msgSize),
-                    mockChannel, mockChannel.profile, remoteAddress, msgSize);
-            } else {
-                // ping message - we just drop all stuff
+            try (BytesStreamOutput output = new ReleasableBytesStreamOutput(msgSize, bigArrays)) {
+                final byte[] buffer = new byte[msgSize];
+                input.readFully(buffer);
+                output.write(minimalHeader);
+                output.writeInt(msgSize);
+                output.write(buffer);
+                final BytesReference bytes = output.bytes();
+                if (TcpTransport.validateMessageHeader(bytes)) {
+                    InetSocketAddress remoteAddress = (InetSocketAddress) socket.getRemoteSocketAddress();
+                    messageReceived(bytes.slice(TcpHeader.MARKER_BYTES_SIZE + TcpHeader.MESSAGE_LENGTH_SIZE, msgSize),
+                        mockChannel, mockChannel.profile, remoteAddress, msgSize);
+                } else {
+                    // ping message - we just drop all stuff
+                }
             }
         }
     }