[improve][broker] Improve Gzip compression, allow excluding specific …

…paths or disabling it (apache#22370)

(cherry picked from commit 15ed659)
(cherry picked from commit cdc5af1)

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -331,6 +331,19 @@ public class ServiceConfiguration implements PulsarConfiguration {
             + "(0 to disable limiting)")
     private int maxHttpServerConnections = 2048;
 
+    @FieldContext(category = CATEGORY_SERVER, doc =
+            "Gzip compression is enabled by default. Specific paths can be excluded from compression.\n"
+                    + "There are 2 syntaxes supported, Servlet url-pattern based, and Regex based.\n"
+                    + "If the spec starts with '^' the spec is assumed to be a regex based path spec and will match "
+                    + "with normal Java regex rules.\n"
+                    + "If the spec starts with '/' then spec is assumed to be a Servlet url-pattern rules path spec "
+                    + "for either an exact match or prefix based match.\n"
+                    + "If the spec starts with '*.' then spec is assumed to be a Servlet url-pattern rules path spec "
+                    + "for a suffix based match.\n"
+                    + "All other syntaxes are unsupported.\n"
+                    + "Disable all compression with ^.* or ^.*$")
+    private List<String> httpServerGzipCompressionExcludedPaths = new ArrayList<>();
+
     @FieldContext(category = CATEGORY_SERVER, doc = "Whether to enable the delayed delivery for messages.")
     private boolean delayedDeliveryEnabled = true;
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/GzipHandlerUtil.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.web;
+
+import java.util.List;
+import org.eclipse.jetty.server.Handler;
+import org.eclipse.jetty.server.handler.gzip.GzipHandler;
+
+public class GzipHandlerUtil {
+    public static Handler wrapWithGzipHandler(Handler innerHandler, List<String> gzipCompressionExcludedPaths) {
+        Handler wrappedHandler;
+        if (isGzipCompressionCompletelyDisabled(gzipCompressionExcludedPaths)) {
+            // no need to add GZIP handler if it's disabled by setting the excluded path to "^.*" or "^.*$"
+            wrappedHandler = innerHandler;
+        } else {
+            // add GZIP handler which is active when the request contains "Accept-Encoding: gzip" header
+            GzipHandler gzipHandler = new GzipHandler();
+            gzipHandler.setHandler(innerHandler);
+            if (gzipCompressionExcludedPaths != null && gzipCompressionExcludedPaths.size() > 0) {
+                gzipHandler.setExcludedPaths(gzipCompressionExcludedPaths.toArray(new String[0]));
+            }
+            wrappedHandler = gzipHandler;
+        }
+        return wrappedHandler;
+    }
+
+    public static boolean isGzipCompressionCompletelyDisabled(List<String> gzipCompressionExcludedPaths) {
+        return gzipCompressionExcludedPaths != null && gzipCompressionExcludedPaths.size() == 1
+                && (gzipCompressionExcludedPaths.get(0).equals("^.*")
+                || gzipCompressionExcludedPaths.get(0).equals("^.*$"));
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java

@@ -44,7 +44,6 @@
 import org.eclipse.jetty.server.handler.RequestLogHandler;
 import org.eclipse.jetty.server.handler.ResourceHandler;
 import org.eclipse.jetty.server.handler.StatisticsHandler;
-import org.eclipse.jetty.server.handler.gzip.GzipHandler;
 import org.eclipse.jetty.servlet.FilterHolder;
 import org.eclipse.jetty.servlet.ServletContextHandler;
 import org.eclipse.jetty.servlet.ServletHolder;
@@ -269,9 +268,7 @@ public void addServlet(String path, ServletHolder servletHolder, boolean require
         }
         filterInitializer.addFilters(servletContextHandler, requiresAuthentication);
 
-        GzipHandler gzipHandler = new GzipHandler();
-        gzipHandler.setHandler(servletContextHandler);
-        handlers.add(gzipHandler);
+        handlers.add(servletContextHandler);
     }
 
     public void addStaticResources(String basePath, String resourcePath) {
@@ -295,8 +292,10 @@ public void start() throws PulsarServerException {
             ContextHandlerCollection contexts = new ContextHandlerCollection();
             contexts.setHandlers(handlers.toArray(new Handler[handlers.size()]));
 
+            Handler handlerForContexts = GzipHandlerUtil.wrapWithGzipHandler(contexts,
+                    pulsar.getConfig().getHttpServerGzipCompressionExcludedPaths());
             HandlerCollection handlerCollection = new HandlerCollection();
-            handlerCollection.setHandlers(new Handler[] { contexts, new DefaultHandler(), requestLogHandler });
+            handlerCollection.setHandlers(new Handler[] {handlerForContexts, new DefaultHandler(), requestLogHandler});
 
             // Metrics handler
             StatisticsHandler stats = new StatisticsHandler();
@@ -307,7 +306,6 @@ public void start() throws PulsarServerException {
             } catch (IllegalArgumentException e) {
                 // Already registered. Eg: in unit tests
             }
-            handlers.add(stats);
 
             server.setHandler(stats);
             server.start();

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java

@@ -158,6 +158,7 @@ protected HttpClient createHttpClient() throws ServletException {
             client.start();
 
             // Content must not be decoded, otherwise the client gets confused.
+            // Allow encoded content, such as "Content-Encoding: gzip", to pass through without decoding it.
             client.getContentDecoderFactories().clear();
 
             // Pass traffic to the client, only intercept what's necessary.