Stop retaining transport responses past serialization (#125163) (#126078)

Remove the `OutboundMessage` class that needlessly holds on to the the response instances after they are not needed any longer. Inlining the logic should save considerably heap under pressure and enabled further optimisations.

backport of #125163

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

@@ -1,60 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the "Elastic License
- * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
- * Public License v 1"; you may not use this file except in compliance with, at
- * your election, the "Elastic License 2.0", the "GNU Affero General Public
- * License v3.0 only", or the "Server Side Public License, v 1".
- */
-package org.elasticsearch.transport;
-
-import org.elasticsearch.TransportVersion;
-import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.common.util.concurrent.ThreadContext;
-
-/**
- * Represents a transport message sent over the network. Subclasses implement serialization and
- * deserialization.
- */
-public abstract class NetworkMessage {
-
-    protected final TransportVersion version;
-    protected final Writeable threadContext;
-    protected final long requestId;
-    protected final byte status;
-    protected final Compression.Scheme compressionScheme;
-
-    NetworkMessage(
-        ThreadContext threadContext,
-        TransportVersion version,
-        byte status,
-        long requestId,
-        Compression.Scheme compressionScheme
-    ) {
-        this.threadContext = threadContext.captureAsWriteable();
-        this.version = version;
-        this.requestId = requestId;
-        this.compressionScheme = adjustedScheme(version, compressionScheme);
-        if (this.compressionScheme != null) {
-            this.status = TransportStatus.setCompress(status);
-        } else {
-            this.status = status;
-        }
-    }
-
-    boolean isCompress() {
-        return TransportStatus.isCompress(status);
-    }
-
-    boolean isHandshake() {
-        return TransportStatus.isHandshake(status);
-    }
-
-    boolean isError() {
-        return TransportStatus.isError(status);
-    }
-
-    private static Compression.Scheme adjustedScheme(TransportVersion version, Compression.Scheme compressionScheme) {
-        return compressionScheme == Compression.Scheme.LZ4 && version.before(Compression.Scheme.LZ4_VERSION) ? null : compressionScheme;
-    }
-}

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

@@ -12,29 +12,39 @@ package org.elasticsearch.transport;
 import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.lucene.store.AlreadyClosedException;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.bytes.CompositeBytesReference;
+import org.elasticsearch.common.bytes.ReleasableBytesReference;
+import org.elasticsearch.common.compress.CompressorFactory;
+import org.elasticsearch.common.io.stream.OutputStreamStreamOutput;
 import org.elasticsearch.common.io.stream.RecyclerBytesStreamOutput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.network.CloseableChannel;
 import org.elasticsearch.common.network.HandlingTimeTracker;
 import org.elasticsearch.common.recycler.Recycler;
 import org.elasticsearch.common.transport.NetworkExceptionHelper;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.RefCounted;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
+import org.elasticsearch.core.Streams;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.IOException;
+import java.util.function.Supplier;
 
 import static org.elasticsearch.core.Strings.format;
 
-final class OutboundHandler {
+public final class OutboundHandler {
 
     private static final Logger logger = LogManager.getLogger(OutboundHandler.class);
 
@@ -83,7 +93,7 @@ final class OutboundHandler {
      *                 thread.
      */
     void sendBytes(TcpChannel channel, BytesReference bytes, ActionListener<Void> listener) {
-        internalSend(channel, bytes, null, listener);
+        internalSend(channel, bytes, () -> "raw bytes", listener);
     }
 
     /**
@@ -102,26 +112,17 @@ final class OutboundHandler {
         final boolean isHandshake
     ) throws IOException, TransportException {
         assert assertValidTransportVersion(transportVersion);
-        final OutboundMessage.Request message = new OutboundMessage.Request(
-            threadPool.getThreadContext(),
-            request,
-            transportVersion,
+        sendMessage(
+            channel,
             action,
+            request,
             requestId,
             isHandshake,
-            compressionScheme
+            compressionScheme,
+            transportVersion,
+            ResponseStatsConsumer.NONE,
+            () -> messageListener.onRequestSent(node, requestId, action, request, options)
         );
-        if (request.tryIncRef() == false) {
-            assert false : "request [" + request + "] has been released already";
-            throw new AlreadyClosedException("request [" + request + "] has been released already");
-        }
-        sendMessage(channel, message, ResponseStatsConsumer.NONE, () -> {
-            try {
-                messageListener.onRequestSent(node, requestId, action, request, options);
-            } finally {
-                request.decRef();
-            }
-        });
     }
 
     /**
@@ -141,23 +142,19 @@ final class OutboundHandler {
         final ResponseStatsConsumer responseStatsConsumer
     ) {
         assert assertValidTransportVersion(transportVersion);
-        OutboundMessage.Response message = new OutboundMessage.Response(
-            threadPool.getThreadContext(),
-            response,
-            transportVersion,
-            requestId,
-            isHandshake,
-            compressionScheme
-        );
-        response.mustIncRef();
+        assert response.hasReferences();
         try {
-            sendMessage(channel, message, responseStatsConsumer, () -> {
-                try {
-                    messageListener.onResponseSent(requestId, action);
-                } finally {
-                    response.decRef();
-                }
-            });
+            sendMessage(
+                channel,
+                null,
+                response,
+                requestId,
+                isHandshake,
+                compressionScheme,
+                transportVersion,
+                responseStatsConsumer,
+                () -> messageListener.onResponseSent(requestId, action)
+            );
         } catch (Exception ex) {
             if (isHandshake) {
                 logger.error(
@@ -187,16 +184,19 @@ final class OutboundHandler {
         final Exception error
     ) {
         assert assertValidTransportVersion(transportVersion);
-        OutboundMessage.Response message = new OutboundMessage.Response(
-            threadPool.getThreadContext(),
-            new RemoteTransportException(nodeName, channel.getLocalAddress(), action, error),
-            transportVersion,
-            requestId,
-            false,
-            null
-        );
+        var msg = new RemoteTransportException(nodeName, channel.getLocalAddress(), action, error);
         try {
-            sendMessage(channel, message, responseStatsConsumer, () -> messageListener.onResponseSent(requestId, action, error));
+            sendMessage(
+                channel,
+                null,
+                msg,
+                requestId,
+                false,
+                null,
+                transportVersion,
+                responseStatsConsumer,
+                () -> messageListener.onResponseSent(requestId, action, error)
+            );
         } catch (Exception sendException) {
             sendException.addSuppressed(error);
             logger.error(() -> format("Failed to send error response on channel [%s], closing channel", channel), sendException);
@@ -206,42 +206,183 @@ final class OutboundHandler {
 
     private void sendMessage(
         TcpChannel channel,
-        OutboundMessage networkMessage,
+        @Nullable String requestAction,
+        Writeable writeable,
+        long requestId,
+        boolean isHandshake,
+        Compression.Scheme compressionScheme,
+        TransportVersion version,
         ResponseStatsConsumer responseStatsConsumer,
         Releasable onAfter
     ) throws IOException {
-        final RecyclerBytesStreamOutput byteStreamOutput;
-        boolean bufferSuccess = false;
-        try {
-            byteStreamOutput = new RecyclerBytesStreamOutput(recycler);
-            bufferSuccess = true;
-        } finally {
-            if (bufferSuccess == false) {
-                Releasables.closeExpectNoException(onAfter);
-            }
-        }
-        final Releasable release = Releasables.wrap(byteStreamOutput, onAfter);
+        compressionScheme = writeable instanceof BytesTransportRequest ? null : compressionScheme;
         final BytesReference message;
         boolean serializeSuccess = false;
+        final boolean isError = writeable instanceof RemoteTransportException;
+        final RecyclerBytesStreamOutput byteStreamOutput = new RecyclerBytesStreamOutput(recycler);
         try {
-            message = networkMessage.serialize(byteStreamOutput);
+            message = serialize(
+                requestAction,
+                requestId,
+                isHandshake,
+                version,
+                isError,
+                compressionScheme,
+                writeable,
+                threadPool.getThreadContext(),
+                byteStreamOutput
+            );
             serializeSuccess = true;
         } catch (Exception e) {
-            logger.warn(() -> "failed to serialize outbound message [" + networkMessage + "]", e);
+            logger.warn(() -> "failed to serialize outbound message [" + writeable + "]", e);
             throw e;
         } finally {
             if (serializeSuccess == false) {
-                release.close();
+                Releasables.close(byteStreamOutput, onAfter);
             }
         }
         responseStatsConsumer.addResponseStats(message.length());
-        internalSend(channel, message, networkMessage, ActionListener.running(release::close));
+        final var responseType = writeable.getClass();
+        final boolean compress = compressionScheme != null;
+        internalSend(
+            channel,
+            message,
+            requestAction == null
+                ? () -> "Response{" + requestId + "}{" + isError + "}{" + compress + "}{" + isHandshake + "}{" + responseType + "}"
+                : () -> "Request{" + requestAction + "}{" + requestId + "}{" + isError + "}{" + compress + "}{" + isHandshake + "}",
+            ActionListener.releasing(
+                message instanceof ReleasableBytesReference r
+                    ? Releasables.wrap(byteStreamOutput, onAfter, r)
+                    : Releasables.wrap(byteStreamOutput, onAfter)
+            )
+        );
+    }
+
+    // public for tests
+    public static BytesReference serialize(
+        @Nullable String requestAction,
+        long requestId,
+        boolean isHandshake,
+        TransportVersion version,
+        boolean isError,
+        Compression.Scheme compressionScheme,
+        Writeable writeable,
+        ThreadContext threadContext,
+        RecyclerBytesStreamOutput byteStreamOutput
+    ) throws IOException {
+        compressionScheme = compressionScheme == Compression.Scheme.LZ4 && version.before(Compression.Scheme.LZ4_VERSION)
+            ? null
+            : compressionScheme;
+        assert byteStreamOutput.position() == 0;
+        byteStreamOutput.setTransportVersion(version);
+        final int headerSize = TcpHeader.headerSize(version);
+        byteStreamOutput.skip(headerSize);
+        final int variableHeaderLength;
+        if (version.onOrAfter(TcpHeader.VERSION_WITH_HEADER_SIZE)) {
+            threadContext.writeTo(byteStreamOutput);
+            if (requestAction != null) {
+                if (version.before(TransportVersions.V_8_0_0)) {
+                    // empty features array
+                    byteStreamOutput.writeStringArray(Strings.EMPTY_ARRAY);
+                }
+                byteStreamOutput.writeString(requestAction);
+            }
+            variableHeaderLength = Math.toIntExact(byteStreamOutput.position() - headerSize);
+        } else {
+            variableHeaderLength = -1;
+        }
+        BytesReference message = serializeMessageBody(
+            writeable,
+            compressionScheme,
+            version,
+            byteStreamOutput,
+            variableHeaderLength,
+            threadContext,
+            requestAction
+        );
+        byte status = 0;
+        if (requestAction == null) {
+            status = TransportStatus.setResponse(status);
+        }
+        if (isHandshake) {
+            status = TransportStatus.setHandshake(status);
+        }
+        if (isError) {
+            status = TransportStatus.setError(status);
+        }
+        if (compressionScheme != null) {
+            status = TransportStatus.setCompress(status);
+        }
+        byteStreamOutput.seek(0);
+        TcpHeader.writeHeader(byteStreamOutput, requestId, status, version, message.length() - headerSize, variableHeaderLength);
+        return message;
+    }
+
+    private static BytesReference serializeMessageBody(
+        Writeable writeable,
+        Compression.Scheme compressionScheme,
+        TransportVersion version,
+        RecyclerBytesStreamOutput byteStreamOutput,
+        int variableHeaderLength,
+        ThreadContext threadContext,
+        String requestAction
+    ) throws IOException {
+        // The compressible bytes stream will not close the underlying bytes stream
+        final StreamOutput stream = compressionScheme != null ? wrapCompressed(compressionScheme, byteStreamOutput) : byteStreamOutput;
+        final ReleasableBytesReference zeroCopyBuffer;
+        try {
+            stream.setTransportVersion(version);
+            if (variableHeaderLength == -1) {
+                threadContext.writeTo(stream);
+                if (requestAction != null) {
+                    stream.writeStringArray(Strings.EMPTY_ARRAY);
+                    stream.writeString(requestAction);
+                }
+            }
+            if (writeable instanceof BytesTransportRequest bRequest) {
+                bRequest.writeThin(stream);
+                zeroCopyBuffer = bRequest.bytes;
+            } else if (writeable instanceof RemoteTransportException remoteTransportException) {
+                stream.writeException(remoteTransportException);
+                zeroCopyBuffer = ReleasableBytesReference.empty();
+            } else {
+                writeable.writeTo(stream);
+                zeroCopyBuffer = ReleasableBytesReference.empty();
+            }
+        } finally {
+            // We have to close here before accessing the bytes when using compression to ensure that some marker bytes (EOS marker)
+            // are written.
+            if (compressionScheme != null) {
+                stream.close();
+            }
+        }
+        final BytesReference msg = byteStreamOutput.bytes();
+        if (zeroCopyBuffer.length() == 0) {
+            return msg;
+        }
+        zeroCopyBuffer.mustIncRef();
+        return new ReleasableBytesReference(CompositeBytesReference.of(msg, zeroCopyBuffer), (RefCounted) zeroCopyBuffer);
+    }
+
+    // compressed stream wrapped bytes must be no-close wrapped since we need to close the compressed wrapper below to release
+    // resources and write EOS marker bytes but must not yet release the bytes themselves
+    private static StreamOutput wrapCompressed(Compression.Scheme compressionScheme, RecyclerBytesStreamOutput bytesStream)
+        throws IOException {
+        if (compressionScheme == Compression.Scheme.DEFLATE) {
+            return new OutputStreamStreamOutput(
+                CompressorFactory.COMPRESSOR.threadLocalOutputStream(org.elasticsearch.core.Streams.noCloseStream(bytesStream))
+            );
+        } else if (compressionScheme == Compression.Scheme.LZ4) {
+            return new OutputStreamStreamOutput(Compression.Scheme.lz4OutputStream(Streams.noCloseStream(bytesStream)));
+        } else {
+            throw new IllegalArgumentException("Invalid compression scheme: " + compressionScheme);
+        }
     }
 
     private void internalSend(
         TcpChannel channel,
         BytesReference reference,
-        @Nullable OutboundMessage message,
+        Supplier<String> messageDescription,
         ActionListener<Void> listener
     ) {
         final long startTime = threadPool.rawRelativeTimeInMillis();
@@ -281,7 +422,7 @@ final class OutboundHandler {
                             logger.warn(
                                 "sending transport message [{}] of size [{}] on [{}] took [{}ms] which is above the warn "
                                     + "threshold of [{}ms] with success [{}]",
-                                message,
+                                messageDescription.get(),
                                 messageSize,
                                 channel,
                                 took,

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

@@ -1,206 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the "Elastic License
- * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
- * Public License v 1"; you may not use this file except in compliance with, at
- * your election, the "Elastic License 2.0", the "GNU Affero General Public
- * License v3.0 only", or the "Server Side Public License, v 1".
- */
-package org.elasticsearch.transport;
-
-import org.elasticsearch.TransportVersion;
-import org.elasticsearch.TransportVersions;
-import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.bytes.BytesArray;
-import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.bytes.CompositeBytesReference;
-import org.elasticsearch.common.compress.CompressorFactory;
-import org.elasticsearch.common.io.stream.OutputStreamStreamOutput;
-import org.elasticsearch.common.io.stream.RecyclerBytesStreamOutput;
-import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.common.util.concurrent.ThreadContext;
-import org.elasticsearch.core.Streams;
-
-import java.io.IOException;
-
-abstract class OutboundMessage extends NetworkMessage {
-
-    protected final Writeable message;
-
-    OutboundMessage(
-        ThreadContext threadContext,
-        TransportVersion version,
-        byte status,
-        long requestId,
-        Compression.Scheme compressionScheme,
-        Writeable message
-    ) {
-        super(threadContext, version, status, requestId, compressionScheme);
-        this.message = message;
-    }
-
-    BytesReference serialize(RecyclerBytesStreamOutput bytesStream) throws IOException {
-        bytesStream.setTransportVersion(version);
-        bytesStream.skip(TcpHeader.headerSize(version));
-
-        // The compressible bytes stream will not close the underlying bytes stream
-        BytesReference reference;
-        int variableHeaderLength = -1;
-        final long preHeaderPosition = bytesStream.position();
-
-        if (version.onOrAfter(TcpHeader.VERSION_WITH_HEADER_SIZE)) {
-            writeVariableHeader(bytesStream);
-            variableHeaderLength = Math.toIntExact(bytesStream.position() - preHeaderPosition);
-        }
-
-        final boolean compress = TransportStatus.isCompress(status);
-        final StreamOutput stream = compress ? wrapCompressed(bytesStream) : bytesStream;
-        final BytesReference zeroCopyBuffer;
-        try {
-            stream.setTransportVersion(version);
-            if (variableHeaderLength == -1) {
-                writeVariableHeader(stream);
-            }
-            if (message instanceof BytesTransportRequest bRequest) {
-                bRequest.writeThin(stream);
-                zeroCopyBuffer = bRequest.bytes;
-            } else if (message instanceof RemoteTransportException) {
-                stream.writeException((RemoteTransportException) message);
-                zeroCopyBuffer = BytesArray.EMPTY;
-            } else {
-                message.writeTo(stream);
-                zeroCopyBuffer = BytesArray.EMPTY;
-            }
-        } finally {
-            // We have to close here before accessing the bytes when using compression to ensure that some marker bytes (EOS marker)
-            // are written.
-            if (compress) {
-                stream.close();
-            }
-        }
-        final BytesReference message = bytesStream.bytes();
-        if (zeroCopyBuffer.length() == 0) {
-            reference = message;
-        } else {
-            reference = CompositeBytesReference.of(message, zeroCopyBuffer);
-        }
-
-        bytesStream.seek(0);
-        final int contentSize = reference.length() - TcpHeader.headerSize(version);
-        TcpHeader.writeHeader(bytesStream, requestId, status, version, contentSize, variableHeaderLength);
-        return reference;
-    }
-
-    // compressed stream wrapped bytes must be no-close wrapped since we need to close the compressed wrapper below to release
-    // resources and write EOS marker bytes but must not yet release the bytes themselves
-    private StreamOutput wrapCompressed(RecyclerBytesStreamOutput bytesStream) throws IOException {
-        if (compressionScheme == Compression.Scheme.DEFLATE) {
-            return new OutputStreamStreamOutput(
-                CompressorFactory.COMPRESSOR.threadLocalOutputStream(org.elasticsearch.core.Streams.noCloseStream(bytesStream))
-            );
-        } else if (compressionScheme == Compression.Scheme.LZ4) {
-            return new OutputStreamStreamOutput(Compression.Scheme.lz4OutputStream(Streams.noCloseStream(bytesStream)));
-        } else {
-            throw new IllegalArgumentException("Invalid compression scheme: " + compressionScheme);
-        }
-    }
-
-    protected void writeVariableHeader(StreamOutput stream) throws IOException {
-        threadContext.writeTo(stream);
-    }
-
-    static class Request extends OutboundMessage {
-
-        private final String action;
-
-        Request(
-            ThreadContext threadContext,
-            Writeable message,
-            TransportVersion version,
-            String action,
-            long requestId,
-            boolean isHandshake,
-            Compression.Scheme compressionScheme
-        ) {
-            super(threadContext, version, setStatus(isHandshake), requestId, adjustCompressionScheme(compressionScheme, message), message);
-            this.action = action;
-        }
-
-        @Override
-        protected void writeVariableHeader(StreamOutput stream) throws IOException {
-            super.writeVariableHeader(stream);
-            if (version.before(TransportVersions.V_8_0_0)) {
-                // empty features array
-                stream.writeStringArray(Strings.EMPTY_ARRAY);
-            }
-            stream.writeString(action);
-        }
-
-        // Do not compress instances of BytesTransportRequest
-        private static Compression.Scheme adjustCompressionScheme(Compression.Scheme compressionScheme, Writeable message) {
-            if (message instanceof BytesTransportRequest) {
-                return null;
-            } else {
-                return compressionScheme;
-            }
-        }
-
-        private static byte setStatus(boolean isHandshake) {
-            byte status = 0;
-            status = TransportStatus.setRequest(status);
-            if (isHandshake) {
-                status = TransportStatus.setHandshake(status);
-            }
-
-            return status;
-        }
-
-        @Override
-        public String toString() {
-            return "Request{" + action + "}{" + requestId + "}{" + isError() + "}{" + isCompress() + "}{" + isHandshake() + "}";
-        }
-    }
-
-    static class Response extends OutboundMessage {
-
-        Response(
-            ThreadContext threadContext,
-            Writeable message,
-            TransportVersion version,
-            long requestId,
-            boolean isHandshake,
-            Compression.Scheme compressionScheme
-        ) {
-            super(threadContext, version, setStatus(isHandshake, message), requestId, compressionScheme, message);
-        }
-
-        private static byte setStatus(boolean isHandshake, Writeable message) {
-            byte status = 0;
-            status = TransportStatus.setResponse(status);
-            if (message instanceof RemoteTransportException) {
-                status = TransportStatus.setError(status);
-            }
-            if (isHandshake) {
-                status = TransportStatus.setHandshake(status);
-            }
-
-            return status;
-        }
-
-        @Override
-        public String toString() {
-            return "Response{"
-                + requestId
-                + "}{"
-                + isError()
-                + "}{"
-                + isCompress()
-                + "}{"
-                + isHandshake()
-                + "}{"
-                + message.getClass()
-                + "}";
-        }
-    }
-}

server/src/test/java/org/elasticsearch/transport/InboundDecoderTests.java

@@ -54,30 +54,34 @@ public class InboundDecoderTests extends ESTestCase {
         } else {
             threadContext.addResponseHeader(headerKey, headerValue);
         }
-        OutboundMessage message;
-        if (isRequest) {
-            message = new OutboundMessage.Request(
-                threadContext,
-                new TestRequest(randomAlphaOfLength(100)),
-                TransportVersion.current(),
-                action,
-                requestId,
-                false,
-                null
-            );
-        } else {
-            message = new OutboundMessage.Response(
-                threadContext,
-                new TestResponse(randomAlphaOfLength(100)),
-                TransportVersion.current(),
-                requestId,
-                false,
-                null
-            );
-        }
 
         try (RecyclerBytesStreamOutput os = new RecyclerBytesStreamOutput(recycler)) {
-            final BytesReference totalBytes = message.serialize(os);
+            final BytesReference totalBytes;
+            if (isRequest) {
+                totalBytes = OutboundHandler.serialize(
+                    action,
+                    requestId,
+                    false,
+                    TransportVersion.current(),
+                    false,
+                    null,
+                    new TestRequest(randomAlphaOfLength(100)),
+                    threadContext,
+                    os
+                );
+            } else {
+                totalBytes = OutboundHandler.serialize(
+                    null,
+                    requestId,
+                    false,
+                    TransportVersion.current(),
+                    false,
+                    null,
+                    new TestResponse(randomAlphaOfLength(100)),
+                    threadContext,
+                    os
+                );
+            }
             int totalHeaderSize = TcpHeader.headerSize(TransportVersion.current()) + totalBytes.getInt(
                 TcpHeader.VARIABLE_HEADER_SIZE_POSITION
             );
@@ -132,18 +136,18 @@ public class InboundDecoderTests extends ESTestCase {
         final TransportVersion preHeaderVariableInt = TransportHandshaker.V7_HANDSHAKE_VERSION;
         final String contentValue = randomAlphaOfLength(100);
         // 8.0 is only compatible with handshakes on a pre-variable int version
-        final OutboundMessage message = new OutboundMessage.Request(
-            threadContext,
-            new TestRequest(contentValue),
-            preHeaderVariableInt,
-            action,
-            requestId,
-            true,
-            compressionScheme
-        );
-
         try (RecyclerBytesStreamOutput os = new RecyclerBytesStreamOutput(recycler)) {
-            final BytesReference totalBytes = message.serialize(os);
+            final BytesReference totalBytes = OutboundHandler.serialize(
+                action,
+                requestId,
+                true,
+                preHeaderVariableInt,
+                false,
+                compressionScheme,
+                new TestRequest(contentValue),
+                threadContext,
+                os
+            );
             int partialHeaderSize = TcpHeader.headerSize(preHeaderVariableInt);
 
             InboundDecoder decoder = new InboundDecoder(recycler);
@@ -189,18 +193,18 @@ public class InboundDecoderTests extends ESTestCase {
         final String headerValue = randomAlphaOfLength(20);
         threadContext.putHeader(headerKey, headerValue);
         TransportVersion handshakeCompat = TransportHandshaker.V7_HANDSHAKE_VERSION;
-        OutboundMessage message = new OutboundMessage.Request(
-            threadContext,
-            new TestRequest(randomAlphaOfLength(100)),
-            handshakeCompat,
-            action,
-            requestId,
-            true,
-            null
-        );
-
         try (RecyclerBytesStreamOutput os = new RecyclerBytesStreamOutput(recycler)) {
-            final BytesReference bytes = message.serialize(os);
+            BytesReference bytes = OutboundHandler.serialize(
+                action,
+                requestId,
+                true,
+                handshakeCompat,
+                false,
+                null,
+                new TestRequest(randomAlphaOfLength(100)),
+                threadContext,
+                os
+            );
             int totalHeaderSize = TcpHeader.headerSize(handshakeCompat);
 
             InboundDecoder decoder = new InboundDecoder(recycler);
@@ -239,19 +243,20 @@ public class InboundDecoderTests extends ESTestCase {
         final String headerKey = randomAlphaOfLength(10);
         final String headerValue = randomAlphaOfLength(20);
         threadContext.putHeader(headerKey, headerValue);
-        OutboundMessage message = new OutboundMessage.Request(
-            threadContext,
-            new TestRequest(randomAlphaOfLength(100)),
-            transportVersion,
-            action,
-            requestId,
-            true,
-            compressionScheme
-        );
 
+        int totalHeaderSize = TcpHeader.headerSize(transportVersion);
         try (RecyclerBytesStreamOutput os = new RecyclerBytesStreamOutput(recycler)) {
-            final BytesReference bytes = message.serialize(os);
-            int totalHeaderSize = TcpHeader.headerSize(transportVersion);
+            final BytesReference bytes = OutboundHandler.serialize(
+                action,
+                requestId,
+                true,
+                transportVersion,
+                false,
+                compressionScheme,
+                new TestRequest(randomAlphaOfLength(100)),
+                threadContext,
+                os
+            );
 
             InboundDecoder decoder = new InboundDecoder(recycler);
             final ArrayList<Object> fragments = new ArrayList<>();
@@ -290,18 +295,19 @@ public class InboundDecoderTests extends ESTestCase {
             ? randomFrom(TransportHandshaker.ALLOWED_HANDSHAKE_VERSIONS)
             : TransportVersionUtils.randomCompatibleVersion(random());
         logger.info("--> version = {}", version);
-        OutboundMessage message = new OutboundMessage.Request(
-            threadContext,
-            new TestRequest(randomAlphaOfLength(100)),
-            version,
-            action,
-            requestId,
-            isHandshake,
-            randomFrom(Compression.Scheme.DEFLATE, Compression.Scheme.LZ4, null)
-        );
 
         try (RecyclerBytesStreamOutput os = new RecyclerBytesStreamOutput(recycler)) {
-            final BytesReference bytes = message.serialize(os);
+            final BytesReference bytes = OutboundHandler.serialize(
+                action,
+                requestId,
+                isHandshake,
+                version,
+                false,
+                randomFrom(Compression.Scheme.DEFLATE, Compression.Scheme.LZ4, null),
+                new TestRequest(randomAlphaOfLength(100)),
+                threadContext,
+                os
+            );
             try (InboundDecoder clientDecoder = new InboundDecoder(recycler, ChannelType.CLIENT)) {
                 IllegalArgumentException e = expectThrows(
                     IllegalArgumentException.class,
@@ -339,17 +345,19 @@ public class InboundDecoderTests extends ESTestCase {
         final var version = isHandshake
             ? randomFrom(TransportHandshaker.ALLOWED_HANDSHAKE_VERSIONS)
             : TransportVersionUtils.randomCompatibleVersion(random());
-        OutboundMessage message = new OutboundMessage.Response(
-            threadContext,
-            new TestResponse(randomAlphaOfLength(100)),
-            version,
-            requestId,
-            isHandshake,
-            randomFrom(Compression.Scheme.DEFLATE, Compression.Scheme.LZ4, null)
-        );
 
         try (RecyclerBytesStreamOutput os = new RecyclerBytesStreamOutput(recycler)) {
-            final BytesReference bytes = message.serialize(os);
+            final BytesReference bytes = OutboundHandler.serialize(
+                null,
+                requestId,
+                isHandshake,
+                version,
+                false,
+                randomFrom(Compression.Scheme.DEFLATE, Compression.Scheme.LZ4, null),
+                new TestRequest(randomAlphaOfLength(100)),
+                threadContext,
+                os
+            );
             try (InboundDecoder decoder = new InboundDecoder(recycler, ChannelType.SERVER)) {
                 final ReleasableBytesReference releasable1 = wrapAsReleasable(bytes);
                 IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> decoder.decode(releasable1, ignored -> {}));
@@ -380,27 +388,38 @@ public class InboundDecoderTests extends ESTestCase {
         } else {
             threadContext.addResponseHeader(headerKey, headerValue);
         }
-        OutboundMessage message;
+        final BytesReference totalBytes;
         TransportMessage transportMessage;
         Compression.Scheme scheme = randomFrom(Compression.Scheme.DEFLATE, Compression.Scheme.LZ4);
-        if (isRequest) {
-            transportMessage = new TestRequest(randomAlphaOfLength(100));
-            message = new OutboundMessage.Request(
-                threadContext,
-                transportMessage,
-                TransportVersion.current(),
-                action,
-                requestId,
-                false,
-                scheme
-            );
-        } else {
-            transportMessage = new TestResponse(randomAlphaOfLength(100));
-            message = new OutboundMessage.Response(threadContext, transportMessage, TransportVersion.current(), requestId, false, scheme);
-        }
 
         try (RecyclerBytesStreamOutput os = new RecyclerBytesStreamOutput(recycler)) {
-            final BytesReference totalBytes = message.serialize(os);
+            if (isRequest) {
+                transportMessage = new TestRequest(randomAlphaOfLength(100));
+                totalBytes = OutboundHandler.serialize(
+                    action,
+                    requestId,
+                    false,
+                    TransportVersion.current(),
+                    false,
+                    scheme,
+                    transportMessage,
+                    threadContext,
+                    os
+                );
+            } else {
+                transportMessage = new TestResponse(randomAlphaOfLength(100));
+                totalBytes = OutboundHandler.serialize(
+                    null,
+                    requestId,
+                    false,
+                    TransportVersion.current(),
+                    false,
+                    scheme,
+                    transportMessage,
+                    threadContext,
+                    os
+                );
+            }
             final BytesStreamOutput out = new BytesStreamOutput();
             transportMessage.writeTo(out);
             final BytesReference uncompressedBytes = out.bytes();
@@ -458,18 +477,18 @@ public class InboundDecoderTests extends ESTestCase {
         final String headerValue = randomAlphaOfLength(20);
         threadContext.putHeader(headerKey, headerValue);
         TransportVersion handshakeCompat = TransportHandshaker.V7_HANDSHAKE_VERSION;
-        OutboundMessage message = new OutboundMessage.Request(
-            threadContext,
-            new TestRequest(randomAlphaOfLength(100)),
-            handshakeCompat,
-            action,
-            requestId,
-            true,
-            Compression.Scheme.DEFLATE
-        );
-
         try (RecyclerBytesStreamOutput os = new RecyclerBytesStreamOutput(recycler)) {
-            final BytesReference bytes = message.serialize(os);
+            final BytesReference bytes = OutboundHandler.serialize(
+                action,
+                requestId,
+                true,
+                handshakeCompat,
+                false,
+                Compression.Scheme.DEFLATE,
+                new TestRequest(randomAlphaOfLength(100)),
+                threadContext,
+                os
+            );
             int totalHeaderSize = TcpHeader.headerSize(handshakeCompat);
 
             InboundDecoder decoder = new InboundDecoder(recycler);
@@ -495,19 +514,19 @@ public class InboundDecoderTests extends ESTestCase {
         String action = "test-request";
         long requestId = randomNonNegativeLong();
         TransportVersion incompatibleVersion = TransportVersionUtils.getPreviousVersion(TransportVersions.MINIMUM_COMPATIBLE);
-        OutboundMessage message = new OutboundMessage.Request(
-            threadContext,
-            new TestRequest(randomAlphaOfLength(100)),
-            incompatibleVersion,
-            action,
-            requestId,
-            false,
-            Compression.Scheme.DEFLATE
-        );
-
         final ReleasableBytesReference releasable1;
         try (RecyclerBytesStreamOutput os = new RecyclerBytesStreamOutput(recycler)) {
-            final BytesReference bytes = message.serialize(os);
+            final BytesReference bytes = OutboundHandler.serialize(
+                action,
+                requestId,
+                false,
+                incompatibleVersion,
+                false,
+                Compression.Scheme.DEFLATE,
+                new TestRequest(randomAlphaOfLength(100)),
+                threadContext,
+                os
+            );
 
             InboundDecoder decoder = new InboundDecoder(recycler);
             final ArrayList<Object> fragments = new ArrayList<>();

server/src/test/java/org/elasticsearch/transport/InboundHandlerTests.java

@@ -171,18 +171,19 @@ public class InboundHandlerTests extends ESTestCase {
         );
         requestHandlers.registerHandler(registry);
         String requestValue = randomAlphaOfLength(10);
-        OutboundMessage.Request request = new OutboundMessage.Request(
-            threadPool.getThreadContext(),
-            new TestRequest(requestValue),
-            TransportVersion.current(),
+        BytesRefRecycler recycler = new BytesRefRecycler(PageCacheRecycler.NON_RECYCLING_INSTANCE);
+        BytesReference fullRequestBytes = OutboundHandler.serialize(
             action,
             requestId,
             false,
-            null
+            TransportVersion.current(),
+            false,
+            null,
+            new TestRequest(requestValue),
+            threadPool.getThreadContext(),
+            new RecyclerBytesStreamOutput(recycler)
         );
 
-        BytesRefRecycler recycler = new BytesRefRecycler(PageCacheRecycler.NON_RECYCLING_INSTANCE);
-        BytesReference fullRequestBytes = request.serialize(new RecyclerBytesStreamOutput(recycler));
         BytesReference requestContent = fullRequestBytes.slice(headerSize, fullRequestBytes.length() - headerSize);
         Header requestHeader = new Header(
             fullRequestBytes.length() - 6,

server/src/test/java/org/elasticsearch/transport/InboundPipelineTests.java

@@ -23,7 +23,6 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.MockPageCacheRecycler;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.Releasable;
-import org.elasticsearch.core.Streams;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.test.ESTestCase;
@@ -109,47 +108,54 @@ public class InboundPipelineTests extends ESTestCase {
                     final MessageData messageData;
                     Exception expectedExceptionClass = null;
 
-                    OutboundMessage message;
-                    if (isRequest) {
-                        if (rarely()) {
-                            messageData = new MessageData(version, requestId, true, compressionScheme, breakThisAction, null);
-                            message = new OutboundMessage.Request(
-                                threadContext,
-                                new TestRequest(value),
-                                version,
-                                breakThisAction,
+                    BytesReference message;
+                    try (RecyclerBytesStreamOutput temporaryOutput = new RecyclerBytesStreamOutput(recycler)) {
+                        if (isRequest) {
+                            if (rarely()) {
+                                messageData = new MessageData(version, requestId, true, compressionScheme, breakThisAction, null);
+                                message = OutboundHandler.serialize(
+                                    breakThisAction,
+                                    requestId,
+                                    false,
+                                    version,
+                                    false,
+                                    compressionScheme,
+                                    new TestRequest(value),
+                                    threadContext,
+                                    temporaryOutput
+                                );
+                                expectedExceptionClass = new CircuitBreakingException("", CircuitBreaker.Durability.PERMANENT);
+                            } else {
+                                messageData = new MessageData(version, requestId, true, compressionScheme, actionName, value);
+                                message = OutboundHandler.serialize(
+                                    actionName,
+                                    requestId,
+                                    false,
+                                    version,
+                                    false,
+                                    compressionScheme,
+                                    new TestRequest(value),
+                                    threadContext,
+                                    temporaryOutput
+                                );
+                            }
+                        } else {
+                            messageData = new MessageData(version, requestId, false, compressionScheme, null, value);
+                            message = OutboundHandler.serialize(
+                                null,
                                 requestId,
                                 false,
-                                compressionScheme
-                            );
-                            expectedExceptionClass = new CircuitBreakingException("", CircuitBreaker.Durability.PERMANENT);
-                        } else {
-                            messageData = new MessageData(version, requestId, true, compressionScheme, actionName, value);
-                            message = new OutboundMessage.Request(
-                                threadContext,
-                                new TestRequest(value),
                                 version,
-                                actionName,
-                                requestId,
                                 false,
-                                compressionScheme
+                                compressionScheme,
+                                new TestResponse(value),
+                                threadContext,
+                                temporaryOutput
                             );
                         }
-                    } else {
-                        messageData = new MessageData(version, requestId, false, compressionScheme, null, value);
-                        message = new OutboundMessage.Response(
-                            threadContext,
-                            new TestResponse(value),
-                            version,
-                            requestId,
-                            false,
-                            compressionScheme
-                        );
-                    }
 
-                    expected.add(new Tuple<>(messageData, expectedExceptionClass));
-                    try (RecyclerBytesStreamOutput temporaryOutput = new RecyclerBytesStreamOutput(recycler)) {
-                        Streams.copy(message.serialize(temporaryOutput).streamInput(), streamOutput, false);
+                        expected.add(new Tuple<>(messageData, expectedExceptionClass));
+                        message.writeTo(streamOutput);
                     }
                 }
 
@@ -222,23 +228,34 @@ public class InboundPipelineTests extends ESTestCase {
             final boolean isRequest = randomBoolean();
             final long requestId = randomNonNegativeLong();
 
-            OutboundMessage message;
+            BytesReference message;
             if (isRequest) {
-                message = new OutboundMessage.Request(
-                    threadContext,
-                    new TestRequest(value),
-                    invalidVersion,
+                message = OutboundHandler.serialize(
                     actionName,
                     requestId,
                     false,
-                    null
+                    invalidVersion,
+                    false,
+                    null,
+                    new TestRequest(value),
+                    threadContext,
+                    streamOutput
                 );
             } else {
-                message = new OutboundMessage.Response(threadContext, new TestResponse(value), invalidVersion, requestId, false, null);
+                message = OutboundHandler.serialize(
+                    null,
+                    requestId,
+                    false,
+                    invalidVersion,
+                    false,
+                    null,
+                    new TestResponse(value),
+                    threadContext,
+                    streamOutput
+                );
             }
 
-            final BytesReference reference = message.serialize(streamOutput);
-            try (ReleasableBytesReference releasable = ReleasableBytesReference.wrap(reference)) {
+            try (ReleasableBytesReference releasable = ReleasableBytesReference.wrap(message)) {
                 expectThrows(IllegalStateException.class, () -> pipeline.handleBytes(new FakeTcpChannel(), releasable));
             }
 
@@ -267,14 +284,33 @@ public class InboundPipelineTests extends ESTestCase {
             final boolean isRequest = randomBoolean();
             final long requestId = randomNonNegativeLong();
 
-            OutboundMessage message;
+            final BytesReference reference;
             if (isRequest) {
-                message = new OutboundMessage.Request(threadContext, new TestRequest(value), version, actionName, requestId, false, null);
+                reference = OutboundHandler.serialize(
+                    actionName,
+                    requestId,
+                    false,
+                    version,
+                    false,
+                    null,
+                    new TestRequest(value),
+                    threadContext,
+                    streamOutput
+                );
             } else {
-                message = new OutboundMessage.Response(threadContext, new TestResponse(value), version, requestId, false, null);
+                reference = OutboundHandler.serialize(
+                    null,
+                    requestId,
+                    false,
+                    version,
+                    false,
+                    null,
+                    new TestResponse(value),
+                    threadContext,
+                    streamOutput
+                );
             }
 
-            final BytesReference reference = message.serialize(streamOutput);
             final int fixedHeaderSize = TcpHeader.headerSize(TransportVersion.current());
             final int variableHeaderSize = reference.getInt(fixedHeaderSize - 4);
             final int totalHeaderSize = fixedHeaderSize + variableHeaderSize;

server/src/test/java/org/elasticsearch/transport/TransportLoggerTests.java

@@ -70,16 +70,17 @@ public class TransportLoggerTests extends ESTestCase {
         BytesRefRecycler recycler = new BytesRefRecycler(PageCacheRecycler.NON_RECYCLING_INSTANCE);
         Compression.Scheme compress = randomFrom(Compression.Scheme.DEFLATE, Compression.Scheme.LZ4, null);
         try (RecyclerBytesStreamOutput bytesStreamOutput = new RecyclerBytesStreamOutput(recycler)) {
-            OutboundMessage.Request request = new OutboundMessage.Request(
-                new ThreadContext(Settings.EMPTY),
-                new EmptyRequest(),
-                TransportVersion.current(),
+            return OutboundHandler.serialize(
                 "internal:test",
                 randomInt(30),
                 false,
-                compress
+                TransportVersion.current(),
+                false,
+                compress,
+                new EmptyRequest(),
+                new ThreadContext(Settings.EMPTY),
+                bytesStreamOutput
             );
-            return request.serialize(bytesStreamOutput);
         }
     }
 }

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

@@ -1,38 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the "Elastic License
- * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
- * Public License v 1"; you may not use this file except in compliance with, at
- * your election, the "Elastic License 2.0", the "GNU Affero General Public
- * License v3.0 only", or the "Server Side Public License, v 1".
- */
-
-package org.elasticsearch.transport;
-
-import org.elasticsearch.TransportVersion;
-import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.io.stream.RecyclerBytesStreamOutput;
-import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.common.util.concurrent.ThreadContext;
-
-import java.io.IOException;
-
-public class TestOutboundRequestMessage extends OutboundMessage.Request {
-    public TestOutboundRequestMessage(
-        ThreadContext threadContext,
-        Writeable message,
-        TransportVersion version,
-        String action,
-        long requestId,
-        boolean isHandshake,
-        Compression.Scheme compressionScheme
-    ) {
-        super(threadContext, message, version, action, requestId, isHandshake, compressionScheme);
-
-    }
-
-    @Override
-    public BytesReference serialize(RecyclerBytesStreamOutput bytesStream) throws IOException {
-        return super.serialize(bytesStream);
-    }
-}

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/netty4/SecurityNetty4ServerTransportAuthenticationTests.java

@@ -38,13 +38,13 @@ import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.BytesRefRecycler;
 import org.elasticsearch.transport.Compression;
 import org.elasticsearch.transport.EmptyRequest;
+import org.elasticsearch.transport.OutboundHandler;
 import org.elasticsearch.transport.ProxyConnectionStrategy;
 import org.elasticsearch.transport.RemoteClusterPortSettings;
 import org.elasticsearch.transport.RemoteClusterService;
 import org.elasticsearch.transport.RemoteConnectionStrategy;
 import org.elasticsearch.transport.RemoteTransportException;
 import org.elasticsearch.transport.SniffConnectionStrategy;
-import org.elasticsearch.transport.TestOutboundRequestMessage;
 import org.elasticsearch.transport.TransportInterceptor;
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.transport.TransportRequestHandler;
@@ -331,18 +331,19 @@ public class SecurityNetty4ServerTransportAuthenticationTests extends ESTestCase
         TransportAddress[] boundRemoteIngressAddresses = remoteSecurityNetty4ServerTransport.boundRemoteIngressAddress().boundAddresses();
         InetSocketAddress remoteIngressTransportAddress = randomFrom(boundRemoteIngressAddresses).address();
         try (Socket socket = new MockSocket(remoteIngressTransportAddress.getAddress(), remoteIngressTransportAddress.getPort())) {
-            TestOutboundRequestMessage message = new TestOutboundRequestMessage(
-                threadPool.getThreadContext(),
-                new EmptyRequest(),
-                TransportVersion.current(),
+            Recycler<BytesRef> recycler = new BytesRefRecycler(PageCacheRecycler.NON_RECYCLING_INSTANCE);
+            RecyclerBytesStreamOutput out = new RecyclerBytesStreamOutput(recycler);
+            BytesReference bytesReference = OutboundHandler.serialize(
                 "internal:whatever",
                 randomNonNegativeLong(),
                 false,
-                randomFrom(Compression.Scheme.DEFLATE, Compression.Scheme.LZ4, null)
+                TransportVersion.current(),
+                false,
+                randomFrom(Compression.Scheme.DEFLATE, Compression.Scheme.LZ4, null),
+                new EmptyRequest(),
+                threadPool.getThreadContext(),
+                out
             );
-            Recycler<BytesRef> recycler = new BytesRefRecycler(PageCacheRecycler.NON_RECYCLING_INSTANCE);
-            RecyclerBytesStreamOutput out = new RecyclerBytesStreamOutput(recycler);
-            BytesReference bytesReference = message.serialize(out);
             socket.getOutputStream().write(Arrays.copyOfRange(bytesReference.array(), 0, bytesReference.length()));
             socket.getOutputStream().flush();