Remote compression scheme default to deflate (elastic#76580)

Currently the cluster.remote.*.transport.compression_scheme setting
defaults to the transport.compression_scheme value. This commit
modifies this to default to deflate (the existing compression scheme
prior to 7.14) when cluster.remote.*.transport.compress is explicitly set.
This will ensure that users do not accidentally change their compression
scheme for 7.x.

Author: Tim-Brooks

docs/reference/modules/remote-clusters.asciidoc

@@ -301,8 +301,11 @@ separately.
   requests to a specific remote cluster. This setting impacts only requests
   sent to the remote cluster. If an inbound request is compressed, {es}
   compresses the response using the same compression scheme. The setting options
-  are `deflate` and `lz4`. If unset, the global `transport.compression_scheme`
-  is used as the fallback setting.
+  are `deflate` and `lz4`. If unset, and
+  `cluster.remote.<cluster_alias>.transport.compress` is explicitly set, defaults
+  to `deflate`. If unset, and `cluster.remote.<cluster_alias>.transport.compress`
+  is not explicitly set, `transport.compression_scheme` is used as the fallback
+  setting.
 
 [discrete]
 [[remote-cluster-sniff-settings]]

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

@@ -121,12 +121,26 @@ public Writeable.Reader<RemoteConnectionInfo.ModeInfo> getReader() {
 
     static ConnectionProfile buildConnectionProfile(String clusterAlias, Settings settings) {
         ConnectionStrategy mode = REMOTE_CONNECTION_MODE.getConcreteSettingForNamespace(clusterAlias).get(settings);
+        final Setting<Compression.Enabled> compressionSetting = RemoteClusterService.REMOTE_CLUSTER_COMPRESS
+            .getConcreteSettingForNamespace(clusterAlias);
+        final Setting<Compression.Scheme> compressionSchemeSetting = RemoteClusterService.REMOTE_CLUSTER_COMPRESSION_SCHEME
+            .getConcreteSettingForNamespace(clusterAlias);
+        boolean compressionSettingExists = compressionSetting.exists(settings);
+        final Compression.Scheme scheme;
+        if (compressionSettingExists) {
+            if (compressionSchemeSetting.exists(settings)) {
+                scheme = compressionSchemeSetting.get(settings);
+            } else {
+                scheme = Compression.Scheme.DEFLATE;
+            }
+        } else {
+            scheme = compressionSchemeSetting.get(settings);
+        }
         ConnectionProfile.Builder builder = new ConnectionProfile.Builder()
             .setConnectTimeout(TransportSettings.CONNECT_TIMEOUT.get(settings))
             .setHandshakeTimeout(TransportSettings.CONNECT_TIMEOUT.get(settings))
-            .setCompressionEnabled(RemoteClusterService.REMOTE_CLUSTER_COMPRESS.getConcreteSettingForNamespace(clusterAlias).get(settings))
-            .setCompressionScheme(RemoteClusterService.REMOTE_CLUSTER_COMPRESSION_SCHEME
-                .getConcreteSettingForNamespace(clusterAlias).get(settings))
+            .setCompressionEnabled(compressionSetting.get(settings))
+            .setCompressionScheme(scheme)
             .setPingInterval(RemoteClusterService.REMOTE_CLUSTER_PING_SCHEDULE.getConcreteSettingForNamespace(clusterAlias).get(settings))
             .addConnections(0, TransportRequestOptions.Type.BULK, TransportRequestOptions.Type.STATE,
                 TransportRequestOptions.Type.RECOVERY, TransportRequestOptions.Type.PING)

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

@@ -75,6 +75,48 @@ public void testChangeInConnectionProfileMeansTheStrategyMustBeRebuilt() {
         assertTrue(first.shouldRebuildConnection(newBuilder.build()));
     }
 
+    public void testCompressionSchemeDefaults() {
+        // Test explicit default
+        Settings.Builder explicitBuilder = Settings.builder();
+        explicitBuilder.put(RemoteConnectionStrategy.REMOTE_CONNECTION_MODE.getConcreteSettingForNamespace("cluster-alias").getKey(),
+            "proxy");
+        explicitBuilder.put(ProxyConnectionStrategy.PROXY_ADDRESS.getConcreteSettingForNamespace("cluster-alias").getKey(),
+            "127.0.0.1:9300");
+        explicitBuilder.put(RemoteClusterService.REMOTE_CLUSTER_COMPRESS.getConcreteSettingForNamespace("cluster-alias").getKey(),
+            randomFrom("true", "indexing_data", "false"));
+        explicitBuilder.put(TransportSettings.TRANSPORT_COMPRESSION_SCHEME.getKey(), "lz4");
+        ConnectionProfile connectionProfileExplicit = FakeConnectionStrategy.buildConnectionProfile("cluster-alias",
+            explicitBuilder.build());
+        assertEquals(Compression.Scheme.DEFLATE, connectionProfileExplicit.getCompressionScheme());
+
+        // Test explicit set
+        Settings.Builder explicit2Builder = Settings.builder();
+        explicit2Builder.put(RemoteConnectionStrategy.REMOTE_CONNECTION_MODE.getConcreteSettingForNamespace("cluster-alias").getKey(),
+            "proxy");
+        explicit2Builder.put(ProxyConnectionStrategy.PROXY_ADDRESS.getConcreteSettingForNamespace("cluster-alias").getKey(),
+            "127.0.0.1:9300");
+        explicit2Builder.put(RemoteClusterService.REMOTE_CLUSTER_COMPRESS.getConcreteSettingForNamespace("cluster-alias").getKey(),
+            randomFrom("true", "indexing_data", "false"));
+        explicit2Builder.put(RemoteClusterService.REMOTE_CLUSTER_COMPRESSION_SCHEME
+            .getConcreteSettingForNamespace("cluster-alias").getKey(), "lz4");
+        explicit2Builder.put(TransportSettings.TRANSPORT_COMPRESSION_SCHEME.getKey(), "deflate");
+        ConnectionProfile connectionProfileExplicit2 = FakeConnectionStrategy.buildConnectionProfile("cluster-alias",
+            explicit2Builder.build());
+        assertEquals(Compression.Scheme.LZ4, connectionProfileExplicit2.getCompressionScheme());
+
+        // Test implicit
+        Settings.Builder implicitBuilder = Settings.builder();
+        implicitBuilder.put(RemoteConnectionStrategy.REMOTE_CONNECTION_MODE.getConcreteSettingForNamespace("cluster-alias").getKey(),
+            "proxy");
+        implicitBuilder.put(ProxyConnectionStrategy.PROXY_ADDRESS.getConcreteSettingForNamespace("cluster-alias").getKey(),
+            "127.0.0.1:9300");
+        implicitBuilder.put(TransportSettings.TRANSPORT_COMPRESS.getKey(), randomFrom("true", "indexing_data", "false"));
+        implicitBuilder.put(TransportSettings.TRANSPORT_COMPRESSION_SCHEME.getKey(), "lz4");
+        ConnectionProfile connectionProfileImplicit = FakeConnectionStrategy.buildConnectionProfile("cluster-alias",
+            implicitBuilder.build());
+        assertEquals(Compression.Scheme.LZ4, connectionProfileImplicit.getCompressionScheme());
+    }
+
     public void testCorrectChannelNumber() {
         String clusterAlias = "cluster-alias";