Always compress based on the settings (elastic#36522)

Tim-Brooks · Tim-Brooks · commit 35ea94fc5374 · 2018-12-12T11:08:02.000-07:00
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 elastic#36399.
diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/get/TransportGetTaskAction.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/get/TransportGetTaskAction.java
@@ -108,7 +108,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.
diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java
@@ -72,11 +72,6 @@ public TransportNodesSnapshotsStatus(Settings settings, ThreadPool threadPool,
         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);
diff --git a/server/src/main/java/org/elasticsearch/action/bulk/BulkAction.java b/server/src/main/java/org/elasticsearch/action/bulk/BulkAction.java
@@ -45,9 +45,6 @@ public BulkRequestBuilder newRequestBuilder(ElasticsearchClient client) {
 
     @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();
     }
 }
diff --git a/server/src/main/java/org/elasticsearch/action/support/nodes/TransportNodesAction.java b/server/src/main/java/org/elasticsearch/action/support/nodes/TransportNodesAction.java
@@ -91,10 +91,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.
      *
@@ -182,7 +178,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];
diff --git a/server/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java b/server/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java
@@ -222,10 +222,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;
@@ -265,7 +261,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;
diff --git a/server/src/main/java/org/elasticsearch/discovery/zen/PublishClusterStateAction.java b/server/src/main/java/org/elasticsearch/discovery/zen/PublishClusterStateAction.java
@@ -74,6 +74,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 {
 
         /**
@@ -282,14 +287,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
@@ -322,12 +322,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
diff --git a/server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayMetaState.java b/server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayMetaState.java
@@ -68,11 +68,6 @@ public ActionFuture<NodesGatewayMetaState> list(String[] nodesIds, @Nullable Tim
         return execute(new Request(nodesIds).timeout(timeout));
     }
 
-    @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);
diff --git a/server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java b/server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java
@@ -95,11 +95,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);
diff --git a/server/src/main/java/org/elasticsearch/indices/recovery/RemoteRecoveryTargetHandler.java b/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())
diff --git a/server/src/main/java/org/elasticsearch/transport/TcpTransport.java b/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);
             }
         }
 
diff --git a/server/src/main/java/org/elasticsearch/transport/TcpTransportChannel.java b/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);
         }
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportMessageListener.java b/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.
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportRequestOptions.java b/server/src/main/java/org/elasticsearch/transport/TransportRequestOptions.java
@@ -24,23 +24,17 @@
 public class TransportRequestOptions {
 
     private final TimeValue timeout;
-    private final boolean compress;
     private final Type type;
 
-    private TransportRequestOptions(TimeValue timeout, boolean compress, Type type) {
+    private TransportRequestOptions(TimeValue timeout, Type type) {
         this.timeout = timeout;
-        this.compress = compress;
         this.type = type;
     }
 
     public TimeValue timeout() {
         return this.timeout;
     }
 
-    public boolean compress() {
-        return this.compress;
-    }
-
     public Type type() {
         return this.type;
     }
@@ -60,15 +54,11 @@ public static Builder builder() {
     }
 
     public static Builder builder(TransportRequestOptions options) {
-        return new Builder()
-                .withTimeout(options.timeout)
-                .withCompress(options.compress)
-                .withType(options.type());
+        return new Builder().withTimeout(options.timeout).withType(options.type());
     }
 
     public static class Builder {
         private TimeValue timeout;
-        private boolean compress;
         private Type type = Type.REG;
 
         private Builder() {
@@ -83,18 +73,13 @@ public Builder withTimeout(TimeValue timeout) {
             return this;
         }
 
-        public Builder withCompress(boolean compress) {
-            this.compress = compress;
-            return this;
-        }
-
         public Builder withType(Type type) {
             this.type = type;
             return this;
         }
 
         public TransportRequestOptions build() {
-            return new TransportRequestOptions(timeout, compress, type);
+            return new TransportRequestOptions(timeout, type);
         }
     }
 }
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportResponseOptions.java b/server/src/main/java/org/elasticsearch/transport/TransportResponseOptions.java
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportService.java b/server/src/main/java/org/elasticsearch/transport/TransportService.java
diff --git a/test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java b/test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java

Original file line number	Diff line number	Diff line change
`@@ -108,7 +108,6 @@ private void runOnNodeWithTaskIfPossible(Task thisTask, GetTaskRequest request,`
`108`	`108`	`if (request.getTimeout() != null) {`
`109`	`109`	`builder.withTimeout(request.getTimeout());`
`110`	`110`	`}`
`111`		`- builder.withCompress(false);`
`112`	`111`	`DiscoveryNode node = clusterService.state().nodes().get(request.getTaskId().getNodeId());`
`113`	`112`	`if (node == null) {`
`114`	`113`	`// Node is no longer part of the cluster! Try and look the task up from the results index.`
Original file line number	Diff line number	Diff line change
`@@ -45,9 +45,6 @@ public BulkRequestBuilder newRequestBuilder(ElasticsearchClient client) {`
`45`	`45`
`46`	`46`	`@Override`
`47`	`47`	`public TransportRequestOptions transportOptions(Settings settings) {`
`48`		`- return TransportRequestOptions.builder()`
`49`		`- .withType(TransportRequestOptions.Type.BULK)`
`50`		`- .withCompress(settings.getAsBoolean("action.bulk.compress", true)`
`51`		`- ).build();`
	`48`	`+ return TransportRequestOptions.builder().withType(TransportRequestOptions.Type.BULK).build();`
`52`	`49`	`}`
`53`	`50`	`}`
Original file line number	Diff line number	Diff line change
`@@ -91,10 +91,6 @@ protected void doExecute(Task task, NodesRequest request, ActionListener<NodesRe`
`91`	`91`	`new AsyncAction(task, request, listener).start();`
`92`	`92`	`}`
`93`	`93`
`94`		`- protected boolean transportCompress() {`
`95`		`- return false;`
`96`		`- }`
`97`		`-`
`98`	`94`	`/**`
`99`	`95`	`* Map the responses into {@code nodeResponseClass} responses and {@link FailedNodeException}s.`
`100`	`96`	`*`
`@@ -182,7 +178,6 @@ void start() {`
`182`	`178`	`if (request.timeout() != null) {`
`183`	`179`	`builder.withTimeout(request.timeout());`
`184`	`180`	`}`
`185`		`- builder.withCompress(transportCompress());`
`186`	`181`	`for (int i = 0; i < nodes.length; i++) {`
`187`	`182`	`final int idx = i;`
`188`	`183`	`final DiscoveryNode node = nodes[i];`