Fixed compression support for h2c protocol (#4944)

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

CHANGELOG.md

@@ -174,6 +174,8 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Fix a bug on handling an invalid array value for point type field #4900([#4900](https://github.com/opensearch-project/OpenSearch/pull/4900))
 - [BUG]: Allow decommission to support delay timeout ([#4930](https://github.com/opensearch-project/OpenSearch/pull/4930))
 - Fix failing test: VerifyVersionConstantsIT ([#4946](https://github.com/opensearch-project/OpenSearch/pull/4946))
+- Fixed compression support for h2c protocol ([#4944](https://github.com/opensearch-project/OpenSearch/pull/4944))
+
 ### Security
 - CVE-2022-25857 org.yaml:snakeyaml DOS vulnerability ([#4341](https://github.com/opensearch-project/OpenSearch/pull/4341))
 

client/rest-high-level/src/test/java/org/opensearch/client/HighLevelRestClientCompressionIT.java

@@ -31,13 +31,22 @@
 
 package org.opensearch.client;
 
+import org.apache.hc.client5.http.classic.methods.HttpGet;
 import org.apache.hc.client5.http.classic.methods.HttpPost;
 import org.apache.hc.client5.http.classic.methods.HttpPut;
+import org.apache.hc.client5.http.entity.GzipCompressingEntity;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.HttpHeaders;
+import org.apache.hc.core5.http.io.entity.StringEntity;
 import org.opensearch.action.search.SearchRequest;
 import org.opensearch.action.search.SearchResponse;
 
 import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
 
 import static org.hamcrest.Matchers.equalTo;
 
@@ -62,4 +71,32 @@ public void testCompressesResponseIfRequested() throws IOException {
         assertEquals(SAMPLE_DOCUMENT, searchResponse.getHits().getHits()[0].getSourceAsString());
     }
 
+    /**
+     * The default CloseableHttpAsyncClient does not support compression out of the box (so that applies to RestClient
+     * and RestHighLevelClient). To check the compression works on both sides, crafting the request using CloseableHttpClient
+     * instead which uses compression by default.
+     */
+    public void testCompressesRequest() throws IOException, URISyntaxException {
+        try (CloseableHttpClient client = HttpClients.custom().build()) {
+            final Node node = client().getNodes().iterator().next();
+            final URI baseUri = new URI(node.getHost().toURI());
+
+            final HttpPut index = new HttpPut(baseUri.resolve("/company/_doc/1"));
+            index.setEntity(new GzipCompressingEntity(new StringEntity(SAMPLE_DOCUMENT, ContentType.APPLICATION_JSON)));
+            try (CloseableHttpResponse response = client.execute(index)) {
+                assertThat(response.getCode(), equalTo(201));
+            }
+
+            final HttpGet refresh = new HttpGet(baseUri.resolve("/_refresh"));
+            try (CloseableHttpResponse response = client.execute(refresh)) {
+                assertThat(response.getCode(), equalTo(200));
+            }
+
+            final HttpPost search = new HttpPost(baseUri.resolve("/_search"));
+            index.setEntity(new GzipCompressingEntity(new StringEntity("{}", ContentType.APPLICATION_JSON)));
+            try (CloseableHttpResponse response = client.execute(search)) {
+                assertThat(response.getCode(), equalTo(200));
+            }
+        }
+    }
 }

modules/transport-netty4/src/main/java/org/opensearch/http/netty4/Netty4HttpServerTransport.java

@@ -413,18 +413,19 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpMessage msg) throws E
                     // If this handler is hit then no upgrade has been attempted and the client is just talking HTTP
                     final ChannelPipeline pipeline = ctx.pipeline();
                     pipeline.addAfter(ctx.name(), "handler", getRequestHandler());
-                    pipeline.replace(this, "aggregator", aggregator);
+                    pipeline.replace(this, "decoder_compress", new HttpContentDecompressor());
 
-                    ch.pipeline().addLast("decoder_compress", new HttpContentDecompressor());
-                    ch.pipeline().addLast("encoder", new HttpResponseEncoder());
+                    pipeline.addAfter("decoder_compress", "aggregator", aggregator);
                     if (handlingSettings.isCompression()) {
-                        ch.pipeline()
-                            .addAfter("aggregator", "encoder_compress", new HttpContentCompressor(handlingSettings.getCompressionLevel()));
+                        pipeline.addAfter(
+                            "aggregator",
+                            "encoder_compress",
+                            new HttpContentCompressor(handlingSettings.getCompressionLevel())
+                        );
                     }
-                    ch.pipeline().addBefore("handler", "request_creator", requestCreator);
-                    ch.pipeline().addBefore("handler", "response_creator", responseCreator);
-                    ch.pipeline()
-                        .addBefore("handler", "pipelining", new Netty4HttpPipeliningHandler(logger, transport.pipeliningMaxEvents));
+                    pipeline.addBefore("handler", "request_creator", requestCreator);
+                    pipeline.addBefore("handler", "response_creator", responseCreator);
+                    pipeline.addBefore("handler", "pipelining", new Netty4HttpPipeliningHandler(logger, transport.pipeliningMaxEvents));
 
                     ctx.fireChannelRead(ReferenceCountUtil.retain(msg));
                 }

server/src/main/resources/org/opensearch/bootstrap/test-framework.policy

@@ -111,6 +111,11 @@ grant codeBase "${codebase.httpcore5}" {
   permission java.net.SocketPermission "*", "connect";
 };
 
+grant codeBase "${codebase.httpclient5}" {
+  // httpclient5 makes socket connections for rest tests
+  permission java.net.SocketPermission "*", "connect";
+};
+
 grant codeBase "${codebase.httpcore-nio}" {
   // httpcore makes socket connections for rest tests
   permission java.net.SocketPermission "*", "connect";