Always compress based on the settings (#36522)

Currently TransportRequestOptions allows specific requests to request
compression. This commit removes this and always compresses based on the
settings. Additionally, it removes TransportResponseOptions as they
are unused.

This closes #36399.

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/get/TransportGetTaskAction.java

@@ -102,7 +102,6 @@ private void runOnNodeWithTaskIfPossible(Task thisTask, GetTaskRequest request,
         if (request.getTimeout() != null) {
             builder.withTimeout(request.getTimeout());
         }
-        builder.withCompress(false);
         DiscoveryNode node = clusterService.state().nodes().get(request.getTaskId().getNodeId());
         if (node == null) {
             // Node is no longer part of the cluster! Try and look the task up from the results index.

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java

@@ -69,11 +69,6 @@ public TransportNodesSnapshotsStatus(ThreadPool threadPool, ClusterService clust
         this.snapshotShardsService = snapshotShardsService;
     }
 
-    @Override
-    protected boolean transportCompress() {
-        return true; // compress since the metadata can become large
-    }
-
     @Override
     protected NodeRequest newNodeRequest(String nodeId, Request request) {
         return new NodeRequest(nodeId, request);

server/src/main/java/org/elasticsearch/action/bulk/BulkAction.java

@@ -39,9 +39,6 @@ public BulkResponse newResponse() {
 
     @Override
     public TransportRequestOptions transportOptions(Settings settings) {
-        return TransportRequestOptions.builder()
-                .withType(TransportRequestOptions.Type.BULK)
-                .withCompress(settings.getAsBoolean("action.bulk.compress", true)
-                ).build();
+        return TransportRequestOptions.builder().withType(TransportRequestOptions.Type.BULK).build();
     }
 }

server/src/main/java/org/elasticsearch/action/support/nodes/TransportNodesAction.java

@@ -82,10 +82,6 @@ protected void doExecute(Task task, NodesRequest request, ActionListener<NodesRe
         new AsyncAction(task, request, listener).start();
     }
 
-    protected boolean transportCompress() {
-        return false;
-    }
-
     /**
      * Map the responses into {@code nodeResponseClass} responses and {@link FailedNodeException}s.
      *
@@ -173,7 +169,6 @@ void start() {
             if (request.timeout() != null) {
                 builder.withTimeout(request.timeout());
             }
-            builder.withCompress(transportCompress());
             for (int i = 0; i < nodes.length; i++) {
                 final int idx = i;
                 final DiscoveryNode node = nodes[i];

server/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java

@@ -212,10 +212,6 @@ protected TasksResponse newResponse(TasksRequest request, AtomicReferenceArray r
      */
     protected abstract void taskOperation(TasksRequest request, OperationTask task, ActionListener<TaskResponse> listener);
 
-    protected boolean transportCompress() {
-        return false;
-    }
-
     private class AsyncAction {
 
         private final TasksRequest request;
@@ -255,7 +251,6 @@ private void start() {
                 if (request.getTimeout() != null) {
                     builder.withTimeout(request.getTimeout());
                 }
-                builder.withCompress(transportCompress());
                 for (int i = 0; i < nodesIds.length; i++) {
                     final String nodeId = nodesIds[i];
                     final int idx = i;

server/src/main/java/org/elasticsearch/cluster/coordination/PublicationTransportHandler.java

@@ -78,6 +78,10 @@ public class PublicationTransportHandler {
     private final AtomicLong fullClusterStateReceivedCount = new AtomicLong();
     private final AtomicLong incompatibleClusterStateDiffReceivedCount = new AtomicLong();
     private final AtomicLong compatibleClusterStateDiffReceivedCount = new AtomicLong();
+    // -> no need to put a timeout on the options here, because we want the response to eventually be received
+    //  and not log an error if it arrives after the timeout
+    private final TransportRequestOptions stateRequestOptions = TransportRequestOptions.builder()
+        .withType(TransportRequestOptions.Type.STATE).build();
 
     public PublicationTransportHandler(TransportService transportService, NamedWriteableRegistry namedWriteableRegistry,
                                        Function<PublishRequest, PublishWithJoinResponse> handlePublishRequest,
@@ -213,7 +217,6 @@ public String toString() {
             @Override
             public void sendApplyCommit(DiscoveryNode destination, ApplyCommitRequest applyCommitRequest,
                                         ActionListener<TransportResponse.Empty> responseActionListener) {
-                TransportRequestOptions options = TransportRequestOptions.builder().withType(TransportRequestOptions.Type.STATE).build();
                 final String actionName;
                 final TransportRequest transportRequest;
                 if (Coordinator.isZen1Node(destination)) {
@@ -223,7 +226,7 @@ public void sendApplyCommit(DiscoveryNode destination, ApplyCommitRequest applyC
                     actionName = COMMIT_STATE_ACTION_NAME;
                     transportRequest = applyCommitRequest;
                 }
-                transportService.sendRequest(destination, actionName, transportRequest, options,
+                transportService.sendRequest(destination, actionName, transportRequest, stateRequestOptions,
                     new TransportResponseHandler<TransportResponse.Empty>() {
 
                         @Override
@@ -254,11 +257,6 @@ private void sendClusterStateToNode(ClusterState clusterState, BytesReference by
                                         ActionListener<PublishWithJoinResponse> responseActionListener, boolean sendDiffs,
                                         Map<Version, BytesReference> serializedStates) {
         try {
-            // -> no need to put a timeout on the options here, because we want the response to eventually be received
-            //  and not log an error if it arrives after the timeout
-            // -> no need to compress, we already compressed the bytes
-            final TransportRequestOptions options = TransportRequestOptions.builder()
-                .withType(TransportRequestOptions.Type.STATE).withCompress(false).build();
             final BytesTransportRequest request = new BytesTransportRequest(bytes, node.getVersion());
             final Consumer<TransportException> transportExceptionHandler = exp -> {
                 if (sendDiffs && exp.unwrapCause() instanceof IncompatibleClusterStateVersionException) {
@@ -304,7 +302,7 @@ public String executor() {
                 actionName = PUBLISH_STATE_ACTION_NAME;
                 transportResponseHandler = publishWithJoinResponseHandler;
             }
-            transportService.sendRequest(node, actionName, request, options, transportResponseHandler);
+            transportService.sendRequest(node, actionName, request, stateRequestOptions, transportResponseHandler);
         } catch (Exception e) {
             logger.warn(() -> new ParameterizedMessage("error sending cluster state to {}", node), e);
             responseActionListener.onFailure(e);

server/src/main/java/org/elasticsearch/discovery/zen/PublishClusterStateAction.java

@@ -76,6 +76,11 @@ public class PublishClusterStateAction {
     public static final String SEND_ACTION_NAME = "internal:discovery/zen/publish/send";
     public static final String COMMIT_ACTION_NAME = "internal:discovery/zen/publish/commit";
 
+    // -> no need to put a timeout on the options, because we want the state response to eventually be received
+    //  and not log an error if it arrives after the timeout
+    private final TransportRequestOptions stateRequestOptions = TransportRequestOptions.builder()
+        .withType(TransportRequestOptions.Type.STATE).build();
+
     public interface IncomingClusterStateListener {
 
         /**
@@ -284,14 +289,9 @@ private void sendClusterStateToNode(final ClusterState clusterState, BytesRefere
                                         final boolean sendDiffs, final Map<Version, BytesReference> serializedStates) {
         try {
 
-            // -> no need to put a timeout on the options here, because we want the response to eventually be received
-            //  and not log an error if it arrives after the timeout
-            // -> no need to compress, we already compressed the bytes
-            TransportRequestOptions options = TransportRequestOptions.builder()
-                .withType(TransportRequestOptions.Type.STATE).withCompress(false).build();
             transportService.sendRequest(node, SEND_ACTION_NAME,
                     new BytesTransportRequest(bytes, node.getVersion()),
-                    options,
+                    stateRequestOptions,
                     new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {
 
                         @Override
@@ -324,12 +324,9 @@ private void sendCommitToNode(final DiscoveryNode node, final ClusterState clust
         try {
             logger.trace("sending commit for cluster state (uuid: [{}], version [{}]) to [{}]",
                 clusterState.stateUUID(), clusterState.version(), node);
-            TransportRequestOptions options = TransportRequestOptions.builder().withType(TransportRequestOptions.Type.STATE).build();
-            // no need to put a timeout on the options here, because we want the response to eventually be received
-            // and not log an error if it arrives after the timeout
             transportService.sendRequest(node, COMMIT_ACTION_NAME,
                     new CommitClusterStateRequest(clusterState.stateUUID()),
-                    options,
+                    stateRequestOptions,
                     new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {
 
                         @Override

server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayMetaState.java

@@ -66,11 +66,6 @@ public ActionFuture<NodesGatewayMetaState> list(String[] nodesIds, @Nullable Tim
         return future;
     }
 
-    @Override
-    protected boolean transportCompress() {
-        return true; // compress since the metadata can become large
-    }
-
     @Override
     protected NodeRequest newNodeRequest(String nodeId, Request request) {
         return new NodeRequest(nodeId);

server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java

@@ -91,11 +91,6 @@ public void list(ShardId shardId, DiscoveryNode[] nodes,
         execute(new Request(shardId, nodes), listener);
     }
 
-    @Override
-    protected boolean transportCompress() {
-        return true; // this can become big...
-    }
-
     @Override
     protected NodeRequest newNodeRequest(String nodeId, Request request) {
         return new NodeRequest(nodeId, request);

server/src/main/java/org/elasticsearch/indices/recovery/RemoteRecoveryTargetHandler.java

@@ -62,12 +62,10 @@ public RemoteRecoveryTargetHandler(long recoveryId, ShardId shardId, TransportSe
         this.recoverySettings = recoverySettings;
         this.onSourceThrottle = onSourceThrottle;
         this.translogOpsRequestOptions = TransportRequestOptions.builder()
-                .withCompress(true)
                 .withType(TransportRequestOptions.Type.RECOVERY)
                 .withTimeout(recoverySettings.internalActionLongTimeout())
                 .build();
         this.fileChunkRequestOptions = TransportRequestOptions.builder()
-                .withCompress(false)  // lucene files are already compressed and therefore compressing this won't really help much so
                 // we are saving the cpu for other things
                 .withType(TransportRequestOptions.Type.RECOVERY)
                 .withTimeout(recoverySettings.internalActionTimeout())

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

@@ -195,7 +195,7 @@ public abstract class TcpTransport extends AbstractLifecycleComponent implements
     // this lock is here to make sure we close this transport and disconnect all the client nodes
     // connections while no connect operations is going on
     private final ReadWriteLock closeLock = new ReentrantReadWriteLock();
-    private final boolean compressResponses;
+    private final boolean compressAllResponses;
     private volatile BoundTransportAddress boundAddress;
     private final String transportName;
 
@@ -220,16 +220,16 @@ public TcpTransport(String transportName, Settings settings,  Version version, T
         this.pageCacheRecycler = pageCacheRecycler;
         this.circuitBreakerService = circuitBreakerService;
         this.namedWriteableRegistry = namedWriteableRegistry;
-        this.compressResponses = Transport.TRANSPORT_TCP_COMPRESS.get(settings);
+        this.compressAllResponses = Transport.TRANSPORT_TCP_COMPRESS.get(settings);
         this.networkService = networkService;
         this.transportName = transportName;
         this.transportLogger = new TransportLogger();
         this.handshaker = new TransportHandshaker(version, threadPool,
             (node, channel, requestId, v) -> sendRequestToChannel(node, channel, requestId,
                 TransportHandshaker.HANDSHAKE_ACTION_NAME, new TransportHandshaker.HandshakeRequest(version),
-                TransportRequestOptions.EMPTY, v, TransportStatus.setHandshake((byte) 0)),
+                TransportRequestOptions.EMPTY, v, false, TransportStatus.setHandshake((byte) 0)),
             (v, features, channel, response, requestId) -> sendResponse(v, features, channel, response, requestId,
-                TransportHandshaker.HANDSHAKE_ACTION_NAME, TransportResponseOptions.EMPTY, TransportStatus.setHandshake((byte) 0)));
+                TransportHandshaker.HANDSHAKE_ACTION_NAME, false, TransportStatus.setHandshake((byte) 0)));
         this.keepAlive = new TransportKeepAlive(threadPool, this::internalSendMessage);
         this.nodeName = Node.NODE_NAME_SETTING.get(settings);
 
@@ -337,11 +337,7 @@ public void sendRequest(long requestId, String action, TransportRequest request,
                 throw new NodeNotConnectedException(node, "connection already closed");
             }
             TcpChannel channel = channel(options.type());
-
-            if (compress) {
-                options = TransportRequestOptions.builder(options).withCompress(true).build();
-            }
-            sendRequestToChannel(this.node, channel, requestId, action, request, options, getVersion(), (byte) 0);
+            sendRequestToChannel(this.node, channel, requestId, action, request, options, getVersion(), compress, (byte) 0);
         }
     }
 
@@ -768,11 +764,11 @@ private boolean canCompress(TransportRequest request) {
 
     private void sendRequestToChannel(final DiscoveryNode node, final TcpChannel channel, final long requestId, final String action,
                                       final TransportRequest request, TransportRequestOptions options, Version channelVersion,
-                                      byte status) throws IOException, TransportException {
+                                      boolean compressRequest, byte status) throws IOException, TransportException {
 
         // only compress if asked and the request is not bytes. Otherwise only
         // the header part is compressed, and the "body" can't be extracted as compressed
-        final boolean compressMessage = options.compress() && canCompress(request);
+        final boolean compressMessage = compressRequest && canCompress(request);
 
         status = TransportStatus.setRequest(status);
         ReleasableBytesStreamOutput bStream = new ReleasableBytesStreamOutput(bigArrays);
@@ -871,8 +867,8 @@ public void sendResponse(
         final TransportResponse response,
         final long requestId,
         final String action,
-        final TransportResponseOptions options) throws IOException {
-        sendResponse(nodeVersion, features, channel, response, requestId, action, options, (byte) 0);
+        final boolean compress) throws IOException {
+        sendResponse(nodeVersion, features, channel, response, requestId, action, compress, (byte) 0);
     }
 
     private void sendResponse(
@@ -882,29 +878,26 @@ private void sendResponse(
         final TransportResponse response,
         final long requestId,
         final String action,
-        TransportResponseOptions options,
+        boolean compress,
         byte status) throws IOException {
-        if (compressResponses && options.compress() == false) {
-            options = TransportResponseOptions.builder(options).withCompress(true).build();
-        }
+        boolean compressMessage = compress || compressAllResponses;
 
         status = TransportStatus.setResponse(status);
         ReleasableBytesStreamOutput bStream = new ReleasableBytesStreamOutput(bigArrays);
-        CompressibleBytesOutputStream stream = new CompressibleBytesOutputStream(bStream, options.compress());
+        CompressibleBytesOutputStream stream = new CompressibleBytesOutputStream(bStream, compressMessage);
         boolean addedReleaseListener = false;
         try {
-            if (options.compress()) {
+            if (compressMessage) {
                 status = TransportStatus.setCompress(status);
             }
             threadPool.getThreadContext().writeTo(stream);
             stream.setVersion(nodeVersion);
             stream.setFeatures(features);
             BytesReference message = buildMessage(requestId, status, nodeVersion, response, stream);
 
-            final TransportResponseOptions finalOptions = options;
             // this might be called in a different thread
             ReleaseListener releaseListener = new ReleaseListener(stream,
-                () -> messageListener.onResponseSent(requestId, action, response, finalOptions));
+                () -> messageListener.onResponseSent(requestId, action, response));
             internalSendMessage(channel, message, releaseListener);
             addedReleaseListener = true;
         } finally {
@@ -1530,9 +1523,9 @@ public void onRequestReceived(long requestId, String action) {
         }
 
         @Override
-        public void onResponseSent(long requestId, String action, TransportResponse response, TransportResponseOptions finalOptions) {
+        public void onResponseSent(long requestId, String action, TransportResponse response) {
             for (TransportMessageListener listener : listeners) {
-                listener.onResponseSent(requestId, action, response, finalOptions);
+                listener.onResponseSent(requestId, action, response);
             }
         }
 

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

@@ -61,13 +61,7 @@ public String getProfileName() {
     @Override
     public void sendResponse(TransportResponse response) throws IOException {
         try {
-            TransportResponseOptions options;
-            if (compressResponse) {
-                options = TransportResponseOptions.builder().withCompress(true).build();
-            } else {
-                options = TransportResponseOptions.EMPTY;
-            }
-            transport.sendResponse(version, features, channel, response, requestId, action, options);
+            transport.sendResponse(version, features, channel, response, requestId, action, compressResponse);
         } finally {
             release(false);
         }

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

@@ -35,9 +35,8 @@ default void onRequestReceived(long requestId, String action) {}
      * @param requestId the request ID (unique per client)
      * @param action the request action
      * @param response the response send
-     * @param finalOptions the response options
      */
-    default void onResponseSent(long requestId, String action, TransportResponse response, TransportResponseOptions finalOptions) {}
+    default void onResponseSent(long requestId, String action, TransportResponse response) {}
 
     /***
      * Called for every failed action response after the response has been passed to the underlying network implementation.