FIX: set default value for enable_compression (#3190)

Signed-off-by: George Chen <[email protected]>
opensearch-project · Aug 18, 2023 · a250bdd · a250bdd
1 parent 6322389
commit a250bdd
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 2 deletions.
diff --git a/data-prepper-plugins/opensearch/README.md b/data-prepper-plugins/opensearch/README.md
@@ -95,7 +95,7 @@ Default is null.
 
 - `index_type` (optional): a String from the list [`custom`, `trace-analytics-raw`, `trace-analytics-service-map`, `management_disabled`], which represents an index type. Defaults to `custom` if `serverless` is `false` in [AWS Configuration](#aws_configuration), otherwise defaults to `management_disabled`. This index_type instructs Sink plugin what type of data it is handling.
 
-- `enable_request_compression` (optional): A boolean that enables or disables request compression when sending requests to OpenSearch. Default is true.
+- `enable_request_compression` (optional): A boolean that enables or disables request compression when sending requests to OpenSearch. For `distribution_version` set to `es6`, default value is `false`, otherwise default value is `true`. 
 
 ```
     APM trace analytics raw span data type example:

diff --git a/...main/java/org/opensearch/dataprepper/plugins/sink/opensearch/ConnectionConfiguration.java b/...main/java/org/opensearch/dataprepper/plugins/sink/opensearch/ConnectionConfiguration.java
@@ -58,6 +58,7 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
+import static org.opensearch.dataprepper.plugins.sink.opensearch.index.IndexConfiguration.DISTRIBUTION_VERSION;
 
 public class ConnectionConfiguration {
   private static final Logger LOG = LoggerFactory.getLogger(OpenSearchSink.class);
@@ -234,7 +235,11 @@ public static ConnectionConfiguration readConnectionConfiguration(final PluginSe
     final String proxy = pluginSetting.getStringOrDefault(PROXY, null);
     builder = builder.withProxy(proxy);
 
-    final boolean requestCompressionEnabled = pluginSetting.getBooleanOrDefault(REQUEST_COMPRESSION_ENABLED, true);
+    final String distributionVersionName = pluginSetting.getStringOrDefault(DISTRIBUTION_VERSION,
+            DistributionVersion.DEFAULT.getVersion());
+    final DistributionVersion distributionVersion = DistributionVersion.fromTypeName(distributionVersionName);
+    final boolean requestCompressionEnabled = pluginSetting.getBooleanOrDefault(
+            REQUEST_COMPRESSION_ENABLED, !DistributionVersion.ES6.equals(distributionVersion));
     builder = builder.withRequestCompressionEnabled(requestCompressionEnabled);
 
     return builder.build();

diff --git a/...java/org/opensearch/dataprepper/plugins/sink/opensearch/ConnectionConfigurationTests.java b/...java/org/opensearch/dataprepper/plugins/sink/opensearch/ConnectionConfigurationTests.java
@@ -50,6 +50,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.opensearch.dataprepper.plugins.sink.opensearch.ConnectionConfiguration.SERVERLESS;
+import static org.opensearch.dataprepper.plugins.sink.opensearch.index.IndexConfiguration.DISTRIBUTION_VERSION;
 
 @ExtendWith(MockitoExtension.class)
 class ConnectionConfigurationTests {
@@ -84,6 +85,18 @@ void testReadConnectionConfigurationDefault() {
         assertNull(connectionConfiguration.getConnectTimeout());
         assertNull(connectionConfiguration.getSocketTimeout());
         assertEquals(TEST_PIPELINE_NAME, connectionConfiguration.getPipelineName());
+        assertTrue(connectionConfiguration.isRequestCompressionEnabled());
+    }
+
+    @Test
+    void testReadConnectionConfigurationES6Default() {
+        final Map<String, Object> configMetadata = generateConfigurationMetadata(
+                TEST_HOSTS, null, null, null, null, true, null, null, null, false);
+        configMetadata.put(DISTRIBUTION_VERSION, "es6");
+        final PluginSetting pluginSetting = getPluginSettingByConfigurationMetadata(configMetadata);
+        final ConnectionConfiguration connectionConfiguration =
+                ConnectionConfiguration.readConnectionConfiguration(pluginSetting);
+        assertFalse(connectionConfiguration.isRequestCompressionEnabled());
     }
 
     @Test