Enable compression for requests to Elasticsearch

v2/elasticsearch-common/src/main/java/com/google/cloud/teleport/v2/elasticsearch/options/ElasticsearchWriteOptions.java

@@ -325,4 +325,14 @@ public interface ElasticsearchWriteOptions extends PipelineOptions {
   Integer getSocketTimeout();
 
   void setSocketTimeout(Integer socketTimeout);
+
+  @TemplateParameter.Boolean(
+      order = 28,
+      optional = true,
+      description = "Enable compression of requests.",
+      helpText = "Whether to compress requests to Elasticsearch. Defaults to: true.")
+  @Default.Boolean(true)
+  Boolean getCompressionEnabled();
+
+  void setCompressionEnabled(Boolean compressionEnabled);
 }

v2/elasticsearch-common/src/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/WriteToElasticsearch.java

@@ -158,6 +158,9 @@ public PDone expand(PCollection<String> jsonStrings) {
     if (options().getSocketTimeout() != null) {
       config = config.withSocketTimeout(options().getSocketTimeout());
     }
+    if (options().getCompressionEnabled() != null) {
+      config = config.withCompressionEnabled(options().getCompressionEnabled());
+    }
 
     ElasticsearchIO.Write elasticsearchWriter =
         ElasticsearchIO.write()

v2/elasticsearch-common/src/main/java/com/google/cloud/teleport/v2/elasticsearch/utils/ElasticsearchIO.java

@@ -272,6 +272,8 @@ public abstract static class ConnectionConfiguration implements Serializable {
 
     public abstract String getUserAgent();
 
+    public abstract boolean isCompressionEnabled();
+
     abstract Builder builder();
 
     @AutoValue.Builder
@@ -304,6 +306,8 @@ abstract static class Builder {
 
       abstract Builder setUserAgent(String userAgent);
 
+      abstract Builder setCompressionEnabled(boolean compressionEnabled);
+
       abstract ConnectionConfiguration build();
     }
 
@@ -330,6 +334,7 @@ public static ConnectionConfiguration create(
           .setTrustSelfSignedCerts(false)
           .setDisableCertificateValidation(false)
           .setUserAgent(userAgent)
+          .setCompressionEnabled(true)
           .build();
     }
 
@@ -485,6 +490,19 @@ public ConnectionConfiguration withConnectTimeout(Integer connectTimeout) {
       return builder().setConnectTimeout(connectTimeout).build();
     }
 
+    /**
+     * Configure whether the REST client should compress requests using gzip content encoding and
+     * add the "Accept-Encoding: gzip". The default is true.
+     *
+     * @param compressionEnabled Whether to compress requests using gzip content encoding and add
+     *     the "Accept-Encoding: gzip"
+     * @return a {@link ConnectionConfiguration} describes a connection configuration to
+     *     Elasticsearch.
+     */
+    public ConnectionConfiguration withCompressionEnabled(boolean compressionEnabled) {
+      return builder().setCompressionEnabled(compressionEnabled).build();
+    }
+
     private void populateDisplayData(DisplayData.Builder builder) {
       builder.add(DisplayData.item("address", getAddresses().toString()));
       builder.add(DisplayData.item("index", getIndex()));
@@ -516,6 +534,9 @@ RestClient createClient() throws IOException {
       }
 
       restClientBuilder.setDefaultHeaders(this.configureDefaultHeaders());
+      if (isCompressionEnabled()) {
+        restClientBuilder.setCompressionEnabled(true);
+      }
 
       if (isDisableCertificateValidation()) {
         try {

v2/elasticsearch-common/src/test/java/com/google/cloud/teleport/v2/elasticsearch/transforms/WriteToElasticsearchTest.java

@@ -229,6 +229,7 @@ public void testElasticsearchExpandMethod() {
     options.setMaxRetryAttempts(3);
     options.setMaxRetryDuration(10L);
     options.setSocketTimeout(1);
+    options.setCompressionEnabled(true);
 
     // Create some dummy input data for testing
     PCollection<String> input = testPipeline.apply(Create.of("json string 1", "json string 2"));