Always compress cluster state on transport layer

We send cluster states over the wire in response to cluster state
requests and when validating join requests. Today there's no special
treatment of these messages so we just materialize the whole cluster
state into a sequence of buffers which can be tens or hundreds of MBs in
a big cluster, and it's quite possible that a really big cluster would
exceed the 2GiB limit on message size.

This commit forces transport compression on these messages even if it's
usually disabled, which means that the materialized cluster state is
always compressed. It tends to compress well, so this gives us a bit
more headroom in terms of maximum cluster size.

Relates #77466
Closes #79906

Author: DaveCTurner

server/src/main/java/org/elasticsearch/action/admin/cluster/state/ClusterStateRequest.java

@@ -19,12 +19,16 @@
 import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
+import org.elasticsearch.transport.AlwaysCompressedRequest;
 
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Map;
 
-public class ClusterStateRequest extends MasterNodeReadRequest<ClusterStateRequest> implements IndicesRequest.Replaceable {
+public class ClusterStateRequest extends MasterNodeReadRequest<ClusterStateRequest>
+    implements
+        IndicesRequest.Replaceable,
+        AlwaysCompressedRequest {
 
     public static final TimeValue DEFAULT_WAIT_FOR_NODE_TIMEOUT = TimeValue.timeValueMinutes(1);
 

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

@@ -10,12 +10,13 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.transport.AlwaysCompressedRequest;
 import org.elasticsearch.transport.TransportRequest;
 
 import java.io.IOException;
 
-public class ValidateJoinRequest extends TransportRequest {
-    private ClusterState state;
+public class ValidateJoinRequest extends TransportRequest implements AlwaysCompressedRequest {
+    private final ClusterState state;
 
     public ValidateJoinRequest(StreamInput in) throws IOException {
         super(in);

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

@@ -0,0 +1,15 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.transport;
+
+/**
+ * A marker interface for requests that should always be compressed because they (or their responses) might get overwhelmingly large
+ * otherwise.
+ */
+public interface AlwaysCompressedRequest {}

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

@@ -13,6 +13,7 @@
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.Version;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
 import org.elasticsearch.common.io.stream.ByteBufferStreamInput;
 import org.elasticsearch.common.io.stream.NamedWriteableAwareStreamInput;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
@@ -131,7 +132,7 @@ private void messageReceived(TcpChannel channel, InboundMessage message, long st
                         if (header.isError()) {
                             handlerResponseError(streamInput, handler);
                         } else {
-                            handleResponse(remoteAddress, streamInput, handler);
+                            handleResponse(remoteAddress, message.getHeader(), streamInput, handler);
                         }
                         // Check the entire message has been read
                         final int nextByte = streamInput.read();
@@ -149,7 +150,7 @@ private void messageReceived(TcpChannel channel, InboundMessage message, long st
                         }
                     } else {
                         assert header.isError() == false;
-                        handleResponse(remoteAddress, EMPTY_STREAM_INPUT, handler);
+                        handleResponse(remoteAddress, message.getHeader(), EMPTY_STREAM_INPUT, handler);
                     }
                 }
             }
@@ -301,6 +302,7 @@ private static void sendErrorResponse(String actionName, TransportChannel transp
 
     private <T extends TransportResponse> void handleResponse(
         InetSocketAddress remoteAddress,
+        final Header header,
         final StreamInput stream,
         final TransportResponseHandler<T> handler
     ) {
@@ -317,6 +319,7 @@ private <T extends TransportResponse> void handleResponse(
             handleException(handler, serializationException);
             return;
         }
+        assert header.isCompressed() || (response instanceof ClusterStateResponse == false);
         final String executor = handler.executor();
         if (ThreadPool.Names.SAME.equals(executor)) {
             doHandleResponse(handler, response);

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

@@ -8,6 +8,7 @@
 package org.elasticsearch.transport;
 
 import org.elasticsearch.Version;
+import org.elasticsearch.cluster.coordination.ValidateJoinRequest;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -36,6 +37,7 @@ abstract class OutboundMessage extends NetworkMessage {
     ) {
         super(threadContext, version, status, requestId, compressionScheme);
         this.message = message;
+        assert isCompress() || (message instanceof ValidateJoinRequest == false);
     }
 
     BytesReference serialize(BytesStreamOutput bytesStream) throws IOException {

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

@@ -266,6 +266,7 @@ public void sendRequest(long requestId, String action, TransportRequest request,
             // is enabled and the request is a RawIndexingDataTransportRequest which indicates it should be
             // compressed.
             final boolean shouldCompress = compress == Compression.Enabled.TRUE
+                || (request instanceof AlwaysCompressedRequest)
                 || (compress == Compression.Enabled.INDEXING_DATA
                     && request instanceof RawIndexingDataTransportRequest
                     && ((RawIndexingDataTransportRequest) request).isRawIndexingData());