Make org.elasticsearch.action.admin.cluster.state.ClusterStateResponse Compress the Cluster State

[original-brownbear]

Unlike publication which uses a org.elasticsearch.transport.BytesTransportRequest that contains the cluster state as a compressed bytes reference, org.elasticsearch.action.admin.cluster.state.ClusterStateResponse does not compress the cluster state.

This is not ideal when it comes to very large cluster states that are requested via the get-cluster-state REST API from another node as these messages can become quite sizable. We should align the behavior here and use the same serialization approach in this message that we use for publication to limit the size of a full cluster state transport message (which will generally compress very well because settings and mappings tend to be duplicated heavily across indices).

relates #77466

[elasticmachine]

Pinging @elastic/es-data-management (Team:Data Management)

[elasticmachine]

Pinging @elastic/es-distributed (Team:Distributed)

[DaveCTurner]

Discussed this in another channel, we think the simplest approach would be to do this using transport compression: if we always compress cluster-state requests then the responses will also always be compressed, which means it'll always be materialised in compressed form before sending, and also this'll avoid compressing it twice if transport compression is in use. Even better, this is already automatically wire-compatible. Something like this looks like the right sort of idea:

diff --git a/server/src/main/java/org/elasticsearch/transport/TcpTransport.java b/server/src/main/java/org/elasticsearch/transport/TcpTransport.java
index 6bed6c7fe61..23905b68ca7 100644
--- a/server/src/main/java/org/elasticsearch/transport/TcpTransport.java
+++ b/server/src/main/java/org/elasticsearch/transport/TcpTransport.java
@@ -268,7 +268,8 @@ public abstract class TcpTransport extends AbstractLifecycleComponent implements
             final boolean shouldCompress = compress == Compression.Enabled.TRUE
                 || (compress == Compression.Enabled.INDEXING_DATA
                     && request instanceof RawIndexingDataTransportRequest
-                    && ((RawIndexingDataTransportRequest) request).isRawIndexingData());
+                    && ((RawIndexingDataTransportRequest) request).isRawIndexingData())
+                || (request instanceof ClusterStateRequest);
             final Compression.Scheme schemeToUse = shouldCompress ? compressionScheme : null;
             outboundHandler.sendRequest(node, channel, requestId, action, request, options, getVersion(), schemeToUse, false);
         }