Introduce ChunkedZipResponse (#109820)

Adds a utility for implementing REST APIs which construct a streaming
(i.e. pretty-much-constant-memory) `.zip` file response as a (pausable)
sequence of `ChunkedRestResponseBodyPart` instances, where each entry in
the `.zip` file is itself a (pausable) sequence of
`ChunkedRestResponseBodyPart` instances.

Relates #104851

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

@@ -53,6 +53,7 @@
 import org.elasticsearch.http.HttpServerTransport;
 import org.elasticsearch.http.netty4.internal.HttpHeadersAuthenticatorUtils;
 import org.elasticsearch.http.netty4.internal.HttpValidator;
+import org.elasticsearch.rest.ChunkedZipResponse;
 import org.elasticsearch.telemetry.tracing.Tracer;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.netty4.AcceptChannelHandler;
@@ -382,7 +383,16 @@ protected boolean isContentAlwaysEmpty(HttpResponse msg) {
                 })
                 .addLast("aggregator", aggregator);
             if (handlingSettings.compression()) {
-                ch.pipeline().addLast("encoder_compress", new HttpContentCompressor(handlingSettings.compressionLevel()));
+                ch.pipeline().addLast("encoder_compress", new HttpContentCompressor(handlingSettings.compressionLevel()) {
+                    @Override
+                    protected Result beginEncode(HttpResponse httpResponse, String acceptEncoding) throws Exception {
+                        if (ChunkedZipResponse.ZIP_CONTENT_TYPE.equals(httpResponse.headers().get("content-type"))) {
+                            return null;
+                        } else {
+                            return super.beginEncode(httpResponse, acceptEncoding);
+                        }
+                    }
+                });
             }
             ch.pipeline()
                 .addLast(