[improve][test] Add integration test for additional servlets (#17450)

tests/docker-images/java-test-plugins/src/main/java/org/apache/pulsar/tests/integration/plugins/RandomAdditionalServlet.java

@@ -0,0 +1,72 @@
+/**
+ * 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.tests.integration.plugins;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import javax.servlet.ServletOutputStream;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlet;
+import org.apache.pulsar.common.configuration.PulsarConfiguration;
+import org.eclipse.jetty.servlet.ServletHolder;
+
+public class RandomAdditionalServlet extends HttpServlet implements AdditionalServlet {
+
+    private int sequenceLength;
+
+    @Override
+    public void loadConfig(PulsarConfiguration pulsarConfiguration) {
+        sequenceLength = Integer.parseInt(
+                pulsarConfiguration.getProperties().getProperty("randomServletSequenceLength"));
+
+    }
+
+    @Override
+    public String getBasePath() {
+        return "/random";
+    }
+
+    @Override
+    public ServletHolder getServletHolder() {
+        return new ServletHolder(this);
+    }
+
+    @Override
+    public void close() {
+
+    }
+
+    @Override
+    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException {
+        resp.setContentType("text/plain");
+        List<Integer> numbers = IntStream.range(0, sequenceLength).boxed()
+                .collect(Collectors.toCollection(ArrayList::new));
+        Collections.shuffle(numbers);
+        String responseBody = numbers.stream().map(Object::toString).collect(Collectors.joining(","));
+        ServletOutputStream output = resp.getOutputStream();
+        output.write(responseBody.getBytes());
+        output.close();
+    }
+}

tests/docker-images/java-test-plugins/src/main/resources/META-INF/services/additional_servlet.yml

@@ -0,0 +1,22 @@
+#
+# 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.
+#
+
+name: random
+description: Additional servlet that generates a random sequence
+additionalServletClass: org.apache.pulsar.tests.integration.plugins.RandomAdditionalServlet

tests/integration/src/test/java/org/apache/pulsar/tests/integration/plugins/TestAdditionalServlets.java

@@ -0,0 +1,98 @@
+/**
+ * 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.tests.integration.plugins;
+
+import org.apache.pulsar.tests.integration.containers.BrokerContainer;
+import org.apache.pulsar.tests.integration.containers.ProxyContainer;
+import org.apache.pulsar.tests.integration.suites.PulsarTestSuite;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.util.Arrays;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+public class TestAdditionalServlets extends PulsarTestSuite {
+
+    private static final String NAME = "random";
+    private static final String PREFIX = "PULSAR_PREFIX_";
+    public static final int SEQUENCE_LENGTH = 7;
+
+    @Override
+    public void setupCluster() throws Exception {
+        brokerEnvs.put(PREFIX + "additionalServlets", NAME);
+        brokerEnvs.put(PREFIX + "additionalServletDirectory", "/pulsar/examples");
+        brokerEnvs.put(PREFIX + "randomServletSequenceLength", "" + SEQUENCE_LENGTH);
+        brokerEnvs.put(PREFIX + "narExtractionDirectory", "/tmp");
+
+        proxyEnvs.put(PREFIX + "additionalServlets", NAME);
+        proxyEnvs.put(PREFIX + "additionalServletDirectory", "/pulsar/examples");
+        proxyEnvs.put(PREFIX + "randomServletSequenceLength", "" + SEQUENCE_LENGTH);
+        proxyEnvs.put(PREFIX + "narExtractionDirectory", "/tmp");
+
+        super.setupCluster();
+    }
+
+    @Test
+    public void testBrokerAdditionalServlet() throws Exception {
+        BrokerContainer broker = getPulsarCluster().getAnyBroker();
+        String host = broker.getHost();
+        Integer httpPort = broker.getMappedPort(BrokerContainer.BROKER_HTTP_PORT);
+
+        testAddress(host, httpPort);
+    }
+
+
+    @Test
+    public void testProxyAdditionalServlet() throws Exception {
+        ProxyContainer proxy = getPulsarCluster().getProxy();
+
+        String host = proxy.getHost();
+        Integer httpPort = proxy.getMappedPort(ProxyContainer.BROKER_HTTP_PORT);
+
+        testAddress(host, httpPort);
+    }
+
+
+
+    private void testAddress(String host, Integer httpPort) throws IOException, InterruptedException, URISyntaxException {
+        ExecutorService executor = null;
+        try {
+            executor = Executors.newSingleThreadExecutor();
+            HttpClient httpClient = HttpClient.newBuilder().followRedirects(HttpClient.Redirect.ALWAYS)
+                    .executor(executor).build();
+            HttpRequest request = HttpRequest.newBuilder()
+                    .uri(new URI("http://" + host + ":" + httpPort + "/" + NAME + "/")).GET().build();
+            String response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()).body();
+            Assert.assertEquals(IntStream.range(0, SEQUENCE_LENGTH).boxed().collect(Collectors.toSet()),
+                    Arrays.stream(response.split(",")).map(Integer::parseInt).collect(Collectors.toSet()));
+        } finally {
+            if (executor != null) {
+                executor.shutdown();
+            }
+        }
+    }
+}

tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterTestBase.java

@@ -34,6 +34,7 @@
 @Slf4j
 public abstract class PulsarClusterTestBase extends PulsarTestBase {
     protected final Map<String, String> brokerEnvs = new HashMap<>();
+    protected final Map<String, String> proxyEnvs = new HashMap<>();
     protected final List<Integer> brokerAdditionalPorts = new LinkedList<>();
 
     @Override
@@ -107,6 +108,7 @@ public void setupCluster(String namePrefix) throws Exception {
         PulsarClusterSpec.PulsarClusterSpecBuilder specBuilder = PulsarClusterSpec.builder()
                 .clusterName(clusterName)
                 .brokerEnvs(brokerEnvs)
+                .proxyEnvs(proxyEnvs)
                 .brokerAdditionalPorts(brokerAdditionalPorts);
 
         setupCluster(beforeSetupCluster(clusterName, specBuilder).build());

tests/integration/src/test/resources/pulsar-plugin.xml

@@ -23,6 +23,7 @@
     <test name="pulsar-plugin-test-suite" preserve-order="true" >
         <classes>
             <class name="org.apache.pulsar.tests.integration.plugins.TestProtocolHandlers" />
+            <class name="org.apache.pulsar.tests.integration.plugins.TestAdditionalServlets" />
         </classes>
     </test>
 </suite>