Enable LZ4 transport compression by default (#76326)

This commit enables LZ4 transport compression by default at the
indexing_data level.

Relates to #73497.

docs/reference/migration/migrate_8_0/transport.asciidoc

@@ -46,4 +46,35 @@ system property, and ensure that all nodes of the same version are running
 exactly the same build. Setting this system property will result in an error
 on startup.
 ====
+
+.Selective transport compression has been enabled by default.
+[%collapsible]
+====
+*Details* +
+Prior to 8.0, transport compression was disabled by default. Starting in 8.0,
+`transport.compress` defaults to `indexing_data`. This configuration means that
+the propagation of raw indexing data will be compressed between nodes.
+
+*Impact* +
+Inter-node transit will get reduced along the indexing path. In some scenarios,
+CPU usage could increase.
+====
+
+.Transport compression defaults to lz4.
+[%collapsible]
+====
+*Details* +
+Prior to 8.0, the `transport.compression_scheme` setting defaulted to `deflate`. Starting in
+8.0,  `transport.compress_scheme` defaults to `lz4`.
+
+Prior to 8.0, the `cluster.remote.<cluster_alias>.transport.compression_scheme`
+setting defaulted to `deflate` when `cluster.remote.<cluster_alias>.transport.compress`
+was explicitly configured. Starting in 8.0,
+`cluster.remote.<cluster_alias>.transport.compression_scheme` will fallback to
+`transport.compression_scheme` by default.
+
+*Impact* +
+This configuration means that transport compression will produce somewhat lower
+compression ratios in exchange for lower CPU load.
+====
 // end::notable-breaking-changes[]

docs/reference/modules/transport.asciidoc

@@ -53,14 +53,14 @@ Set to `true`, `indexing_data`, or `false` to configure transport compression
 between nodes. The option `true` will compress all data. The option
 `indexing_data` will compress only the raw index data sent between nodes during
 ingest, ccr following (excluding bootstrap), and operations based shard recovery
-(excluding transferring lucene files). Defaults to `false`.
+(excluding transferring lucene files). Defaults to `indexing_data`.
 
 `transport.compression_scheme`::
 (<<static-cluster-setting,Static>>)
 Configures the compression scheme for `transport.compress`. The options are
 `deflate` or `lz4`. If `lz4` is configured and the remote node has not been
 upgraded to a version supporting `lz4`, the traffic will be sent uncompressed.
-Defaults to `deflate`.
+Defaults to `lz4`.
 
 `transport.ping_schedule`::
 (<<static-cluster-setting,Static>>)
@@ -176,16 +176,12 @@ must not be used.
 [[request-compression]]
 ===== Request compression
 
-By default, the `transport.compress` setting is `false` and network-level
-request compression is disabled between nodes in the cluster. This default
-normally makes sense for local cluster communication as compression has a
-noticeable CPU cost and local clusters tend to be set up with fast network
-connections between nodes.
-
-The `transport.compress` configuration option `indexing_data` will only
+The default `transport.compress` configuration option `indexing_data` will only
 compress requests that relate to the transport of raw indexing source data
 between nodes. This option primarily compresses data sent during ingest,
-ccr, and shard recovery.
+ccr, and shard recovery. This default normally makes sense for local cluster
+communication as compressing raw documents tends significantly reduce inter-node
+network usage with minimal CPU impact.
 
 The `transport.compress` setting always configures local cluster request
 compression and is the fallback setting for remote cluster request compression.

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

@@ -51,10 +51,9 @@ public final class TransportSettings {
     public static final Setting.AffixSetting<Integer> PUBLISH_PORT_PROFILE = affixKeySetting("transport.profiles.", "publish_port",
         key -> intSetting(key, -1, -1, Setting.Property.NodeScope));
     public static final Setting<Compression.Enabled> TRANSPORT_COMPRESS =
-        enumSetting(Compression.Enabled.class, "transport.compress", Compression.Enabled.FALSE, Setting.Property.NodeScope);
+        enumSetting(Compression.Enabled.class, "transport.compress", Compression.Enabled.INDEXING_DATA, Setting.Property.NodeScope);
     public static final Setting<Compression.Scheme> TRANSPORT_COMPRESSION_SCHEME =
-        enumSetting(Compression.Scheme.class, "transport.compression_scheme", Compression.Scheme.DEFLATE,
-            Setting.Property.NodeScope);
+        enumSetting(Compression.Scheme.class, "transport.compression_scheme", Compression.Scheme.LZ4, Setting.Property.NodeScope);
     // the scheduled internal ping interval setting, defaults to disabled (-1)
     public static final Setting<TimeValue> PING_SCHEDULE =
         timeSetting("transport.ping_schedule", TimeValue.timeValueSeconds(-1), Setting.Property.NodeScope);

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

@@ -370,11 +370,11 @@ public class RemoteClusterServiceTests extends ESTestCase {
                     TimeValue pingSchedule = TimeValue.timeValueSeconds(randomIntBetween(6, 8));
                     settingsChange.put("cluster.remote.cluster_1.transport.ping_schedule", pingSchedule);
                     boolean compressionScheme = randomBoolean();
-                    Compression.Enabled enabled = randomFrom(Compression.Enabled.TRUE, Compression.Enabled.INDEXING_DATA);
+                    Compression.Enabled enabledChange = randomFrom(Compression.Enabled.TRUE, Compression.Enabled.FALSE);
                     if (compressionScheme) {
-                        settingsChange.put("cluster.remote.cluster_1.transport.compression_scheme", Compression.Scheme.LZ4);
+                        settingsChange.put("cluster.remote.cluster_1.transport.compression_scheme", Compression.Scheme.DEFLATE);
                     } else {
-                        settingsChange.put("cluster.remote.cluster_1.transport.compress", enabled);
+                        settingsChange.put("cluster.remote.cluster_1.transport.compress", enabledChange);
                     }
                     settingsChange.putList("cluster.remote.cluster_1.seeds", cluster1Seed.getAddress().toString());
                     service.validateAndUpdateRemoteCluster("cluster_1", settingsChange.build());
@@ -384,11 +384,11 @@ public class RemoteClusterServiceTests extends ESTestCase {
                     ConnectionProfile connectionProfile = remoteClusterConnection.getConnectionManager().getConnectionProfile();
                     assertEquals(pingSchedule, connectionProfile.getPingInterval());
                     if (compressionScheme) {
-                        assertEquals(Compression.Enabled.FALSE, connectionProfile.getCompressionEnabled());
-                        assertEquals(Compression.Scheme.LZ4, connectionProfile.getCompressionScheme());
-                    } else {
-                        assertEquals(enabled, connectionProfile.getCompressionEnabled());
+                        assertEquals(Compression.Enabled.INDEXING_DATA, connectionProfile.getCompressionEnabled());
                         assertEquals(Compression.Scheme.DEFLATE, connectionProfile.getCompressionScheme());
+                    } else {
+                        assertEquals(enabledChange, connectionProfile.getCompressionEnabled());
+                        assertEquals(Compression.Scheme.LZ4, connectionProfile.getCompressionScheme());
                     }
                 }
             }

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

@@ -45,8 +45,8 @@ public class RemoteConnectionStrategyTests extends ESTestCase {
     public void testChangeInConnectionProfileMeansTheStrategyMustBeRebuilt() {
         ClusterConnectionManager connectionManager = new ClusterConnectionManager(TestProfiles.LIGHT_PROFILE, mock(Transport.class));
         assertEquals(TimeValue.MINUS_ONE, connectionManager.getConnectionProfile().getPingInterval());
-        assertEquals(Compression.Enabled.FALSE, connectionManager.getConnectionProfile().getCompressionEnabled());
-        assertEquals(Compression.Scheme.DEFLATE, connectionManager.getConnectionProfile().getCompressionScheme());
+        assertEquals(Compression.Enabled.INDEXING_DATA, connectionManager.getConnectionProfile().getCompressionEnabled());
+        assertEquals(Compression.Scheme.LZ4, connectionManager.getConnectionProfile().getCompressionScheme());
         RemoteConnectionManager remoteConnectionManager = new RemoteConnectionManager("cluster-alias", connectionManager);
         FakeConnectionStrategy first = new FakeConnectionStrategy("cluster-alias", mock(TransportService.class), remoteConnectionManager,
             RemoteConnectionStrategy.ConnectionStrategy.PROXY);
@@ -63,11 +63,11 @@ public class RemoteConnectionStrategyTests extends ESTestCase {
                 TimeValue.timeValueSeconds(5));
         } else if (change.equals(compress)) {
             newBuilder.put(RemoteClusterService.REMOTE_CLUSTER_COMPRESS.getConcreteSettingForNamespace("cluster-alias").getKey(),
-                randomFrom(Compression.Enabled.INDEXING_DATA, Compression.Enabled.TRUE));
+                randomFrom(Compression.Enabled.FALSE, Compression.Enabled.TRUE));
         } else if (change.equals(compressionScheme)) {
             newBuilder.put(
                 RemoteClusterService.REMOTE_CLUSTER_COMPRESSION_SCHEME.getConcreteSettingForNamespace("cluster-alias").getKey(),
-                Compression.Scheme.LZ4
+                Compression.Scheme.DEFLATE
             );
         } else {
             throw new AssertionError("Unexpected option: " + change);