HTTP API calls hang with 'Accept-Encoding: zstd' (#17408) (#17416)

Signed-off-by: Andriy Redko <[email protected]>
(cherry picked from commit ca8e4f8)

CHANGELOG.md

@@ -15,6 +15,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Removed
 
 ### Fixed
+- Fix HTTP API calls that hang with 'Accept-Encoding: zstd' ([#17408](https://github.com/opensearch-project/OpenSearch/pull/17408))
 
 ### Security
 

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

@@ -61,6 +61,8 @@
 
 import java.net.InetSocketAddress;
 import java.net.SocketOption;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import io.netty.bootstrap.ServerBootstrap;
@@ -75,6 +77,12 @@
 import io.netty.channel.RecvByteBufAllocator;
 import io.netty.channel.socket.nio.NioChannelOption;
 import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.compression.Brotli;
+import io.netty.handler.codec.compression.CompressionOptions;
+import io.netty.handler.codec.compression.DeflateOptions;
+import io.netty.handler.codec.compression.GzipOptions;
+import io.netty.handler.codec.compression.StandardCompressionOptions;
+import io.netty.handler.codec.compression.ZstdEncoder;
 import io.netty.handler.codec.http.HttpContentCompressor;
 import io.netty.handler.codec.http.HttpContentDecompressor;
 import io.netty.handler.codec.http.HttpObjectAggregator;
@@ -374,7 +382,11 @@ protected void initChannel(Channel ch) throws Exception {
             aggregator.setMaxCumulationBufferComponents(transport.maxCompositeBufferComponents);
             ch.pipeline().addLast("aggregator", aggregator);
             if (handlingSettings.isCompression()) {
-                ch.pipeline().addLast("encoder_compress", new HttpContentCompressor(handlingSettings.getCompressionLevel()));
+                ch.pipeline()
+                    .addLast(
+                        "encoder_compress",
+                        new HttpContentCompressor(defaultCompressionOptions(handlingSettings.getCompressionLevel()))
+                    );
             }
             ch.pipeline().addLast("request_creator", requestCreator);
             ch.pipeline().addLast("response_creator", responseCreator);
@@ -427,4 +439,59 @@ protected ChannelInboundHandlerAdapter createHeaderVerifier() {
     protected ChannelInboundHandlerAdapter createDecompressor() {
         return new HttpContentDecompressor();
     }
+
+    /**
+     * Copy of {@link HttpContentCompressor} default compression options with ZSTD excluded:
+     * although zstd-jni is on the classpath, {@link ZstdEncoder} requires direct buffers support
+     * which by default {@link NettyAllocator} does not provide.
+     *
+     * @param compressionLevel
+     *        {@code 1} yields the fastest compression and {@code 9} yields the
+     *        best compression.  {@code 0} means no compression.  The default
+     *        compression level is {@code 6}.
+     *
+     * @return default compression options
+     */
+    private static CompressionOptions[] defaultCompressionOptions(int compressionLevel) {
+        return defaultCompressionOptions(compressionLevel, 15, 8);
+    }
+
+    /**
+     * Copy of {@link HttpContentCompressor} default compression options with ZSTD excluded:
+     * although zstd-jni is on the classpath, {@link ZstdEncoder} requires direct buffers support
+     * which by default {@link NettyAllocator} does not provide.
+     *
+     * @param compressionLevel
+     *        {@code 1} yields the fastest compression and {@code 9} yields the
+     *        best compression.  {@code 0} means no compression.  The default
+     *        compression level is {@code 6}.
+     * @param windowBits
+     *        The base two logarithm of the size of the history buffer.  The
+     *        value should be in the range {@code 9} to {@code 15} inclusive.
+     *        Larger values result in better compression at the expense of
+     *        memory usage.  The default value is {@code 15}.
+     * @param memLevel
+     *        How much memory should be allocated for the internal compression
+     *        state.  {@code 1} uses minimum memory and {@code 9} uses maximum
+     *        memory.  Larger values result in better and faster compression
+     *        at the expense of memory usage.  The default value is {@code 8}
+     *
+     * @return default compression options
+     */
+    private static CompressionOptions[] defaultCompressionOptions(int compressionLevel, int windowBits, int memLevel) {
+        final List<CompressionOptions> options = new ArrayList<CompressionOptions>(4);
+        final GzipOptions gzipOptions = StandardCompressionOptions.gzip(compressionLevel, windowBits, memLevel);
+        final DeflateOptions deflateOptions = StandardCompressionOptions.deflate(compressionLevel, windowBits, memLevel);
+
+        options.add(gzipOptions);
+        options.add(deflateOptions);
+        options.add(StandardCompressionOptions.snappy());
+
+        if (Brotli.isAvailable()) {
+            options.add(StandardCompressionOptions.brotli());
+        }
+
+        return options.toArray(new CompressionOptions[0]);
+    }
+
 }

modules/transport-netty4/src/test/java/org/opensearch/http/netty4/Netty4HttpServerTransportTests.java

@@ -393,7 +393,10 @@ public void dispatchBadRequest(final RestChannel channel, final ThreadContext th
 
             try (Netty4HttpClient client = new Netty4HttpClient()) {
                 DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, url);
-                request.headers().add(HttpHeaderNames.ACCEPT_ENCODING, randomFrom("deflate", "gzip"));
+                // ZSTD is not supported at the moment by NettyAllocator (needs direct buffers),
+                // and Brotly is not on classpath.
+                final String contentEncoding = randomFrom("deflate", "gzip", "snappy", "br", "zstd");
+                request.headers().add(HttpHeaderNames.ACCEPT_ENCODING, contentEncoding);
                 long numOfHugeAllocations = getHugeAllocationCount();
                 final FullHttpResponse response = client.send(remoteAddress.address(), request);
                 try {