You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/08/02 16:26:03 UTC
[pulsar] branch master updated: [feature][pulsar-broker]: Add
additional servlet support to broker (#11498)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 0a68069 [feature][pulsar-broker]: Add additional servlet support to broker (#11498)
0a68069 is described below
commit 0a680695fb6b709f06faa4b780eddc4ba2e50784
Author: Kai Wang <wa...@gmail.com>
AuthorDate: Tue Aug 3 00:25:22 2021 +0800
[feature][pulsar-broker]: Add additional servlet support to broker (#11498)
### Motivation
Currently the broker does not support additional servlet, but some plugins need to add custom web routes on the broker.
### Modifications
Add additional servlet support to broker
---
.../apache/pulsar/broker/ServiceConfiguration.java | 14 ++
.../broker/web/plugin/servlet/package-info.java | 2 +-
.../org/apache/pulsar/broker/PulsarService.java | 181 ++++++++++++-------
.../AdditionalServletWithPulsarService.java | 9 +-
.../broker/web/plugin/servlet/package-info.java | 3 +-
.../pulsar/broker/BrokerAdditionalServletTest.java | 200 +++++++++++++++++++++
.../broker/auth/MockedPulsarServiceBaseTest.java | 5 +
.../AdditionalServletWithPulsarServiceTest.java | 85 +++++++++
.../MockAdditionalServletWithClassLoader.java | 35 +++-
9 files changed, 465 insertions(+), 69 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 08024a9..7894214 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -93,6 +93,8 @@ public class ServiceConfiguration implements PulsarConfiguration {
private static final String CATEGORY_TRANSACTION = "Transaction";
@Category
private static final String CATEGORY_PACKAGES_MANAGEMENT = "Packages Management";
+ @Category
+ private static final String CATEGORY_PLUGIN = "Broker Plugin";
/***** --- pulsar configuration --- ****/
@FieldContext(
@@ -2245,6 +2247,18 @@ public class ServiceConfiguration implements PulsarConfiguration {
/* packages management service configurations (end) */
+ @FieldContext(
+ category = CATEGORY_PLUGIN,
+ doc = "The directory to locate broker additional servlet"
+ )
+ private String additionalServletDirectory = "./brokerAdditionalServlet";
+
+ @FieldContext(
+ category = CATEGORY_PLUGIN,
+ doc = "List of broker additional servlet to load, which is a list of broker additional servlet names"
+ )
+ private Set<String> additionalServlets = Sets.newTreeSet();
+
/**
* @deprecated See {@link #getConfigurationStoreServers}
*/
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java
index 2807dc4..8c3b3b2 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java
@@ -18,6 +18,6 @@
*/
/**
- * Pulsar proxy servlet plugin.
+ * Pulsar additional servlet plugin.
*/
package org.apache.pulsar.broker.web.plugin.servlet;
\ No newline at end of file
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 84395a0..5098c9f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -36,8 +36,10 @@ import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
+import java.net.MalformedURLException;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
@@ -60,6 +62,8 @@ import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.function.Supplier;
+import javax.servlet.ServletException;
+import javax.websocket.DeploymentException;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.Setter;
@@ -116,10 +120,15 @@ import org.apache.pulsar.broker.transaction.pendingack.TransactionPendingAckStor
import org.apache.pulsar.broker.transaction.pendingack.impl.MLPendingAckStore;
import org.apache.pulsar.broker.validator.MultipleListenerValidator;
import org.apache.pulsar.broker.web.WebService;
+import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlet;
+import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServletWithClassLoader;
+import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServletWithPulsarService;
+import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlets;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.transaction.TransactionBufferClient;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
@@ -242,6 +251,7 @@ public class PulsarService implements AutoCloseable {
private HashedWheelTimer transactionTimer;
private BrokerInterceptor brokerInterceptor;
+ private AdditionalServlets brokerAdditionalServlets;
// packages management service
private PackagesManagement packagesManagement;
@@ -380,6 +390,11 @@ public class PulsarService implements AutoCloseable {
this.webSocketService.close();
}
+ if (brokerAdditionalServlets != null) {
+ brokerAdditionalServlets.close();
+ brokerAdditionalServlets = null;
+ }
+
GracefulExecutorServicesShutdown executorServicesShutdown =
GracefulExecutorServicesShutdown
.initiate()
@@ -662,27 +677,10 @@ public class PulsarService implements AutoCloseable {
this.brokerInterceptor.initialize(this);
brokerService.start();
+ // Load additional servlets
+ this.brokerAdditionalServlets = AdditionalServlets.load(config);
+
this.webService = new WebService(this);
- Map<String, Object> attributeMap = Maps.newHashMap();
- attributeMap.put(WebService.ATTRIBUTE_PULSAR_NAME, this);
- Map<String, Object> vipAttributeMap = Maps.newHashMap();
- vipAttributeMap.put(VipStatus.ATTRIBUTE_STATUS_FILE_PATH, this.config.getStatusFilePath());
- vipAttributeMap.put(VipStatus.ATTRIBUTE_IS_READY_PROBE, (Supplier<Boolean>) () -> {
- // Ensure the VIP status is only visible when the broker is fully initialized
- return state == State.Started;
- });
- this.webService.addRestResources("/",
- VipStatus.class.getPackage().getName(), false, vipAttributeMap);
- this.webService.addRestResources("/",
- "org.apache.pulsar.broker.web", false, attributeMap);
- this.webService.addRestResources("/admin",
- "org.apache.pulsar.broker.admin.v1", true, attributeMap);
- this.webService.addRestResources("/admin/v2",
- "org.apache.pulsar.broker.admin.v2", true, attributeMap);
- this.webService.addRestResources("/admin/v3",
- "org.apache.pulsar.broker.admin.v3", true, attributeMap);
- this.webService.addRestResources("/lookup",
- "org.apache.pulsar.broker.lookup", true, attributeMap);
this.metricsServlet = new PrometheusMetricsServlet(
this, config.isExposeTopicLevelMetricsInPrometheus(),
config.isExposeConsumerLevelMetricsInPrometheus(),
@@ -692,46 +690,8 @@ public class PulsarService implements AutoCloseable {
this.pendingMetricsProviders = null;
}
- this.webService.addServlet("/metrics",
- new ServletHolder(metricsServlet),
- false, attributeMap);
-
- if (config.isWebSocketServiceEnabled()) {
- // Use local broker address to avoid different IP address when using a VIP for service discovery
- this.webSocketService = new WebSocketService(null, config);
- this.webSocketService.start();
-
- final WebSocketServlet producerWebSocketServlet = new WebSocketProducerServlet(webSocketService);
- this.webService.addServlet(WebSocketProducerServlet.SERVLET_PATH,
- new ServletHolder(producerWebSocketServlet), true, attributeMap);
- this.webService.addServlet(WebSocketProducerServlet.SERVLET_PATH_V2,
- new ServletHolder(producerWebSocketServlet), true, attributeMap);
-
- final WebSocketServlet consumerWebSocketServlet = new WebSocketConsumerServlet(webSocketService);
- this.webService.addServlet(WebSocketConsumerServlet.SERVLET_PATH,
- new ServletHolder(consumerWebSocketServlet), true, attributeMap);
- this.webService.addServlet(WebSocketConsumerServlet.SERVLET_PATH_V2,
- new ServletHolder(consumerWebSocketServlet), true, attributeMap);
-
- final WebSocketServlet readerWebSocketServlet = new WebSocketReaderServlet(webSocketService);
- this.webService.addServlet(WebSocketReaderServlet.SERVLET_PATH,
- new ServletHolder(readerWebSocketServlet), true, attributeMap);
- this.webService.addServlet(WebSocketReaderServlet.SERVLET_PATH_V2,
- new ServletHolder(readerWebSocketServlet), true, attributeMap);
-
- final WebSocketServlet pingPongWebSocketServlet = new WebSocketPingPongServlet(webSocketService);
- this.webService.addServlet(WebSocketPingPongServlet.SERVLET_PATH,
- new ServletHolder(pingPongWebSocketServlet), true, attributeMap);
- this.webService.addServlet(WebSocketPingPongServlet.SERVLET_PATH_V2,
- new ServletHolder(pingPongWebSocketServlet), true, attributeMap);
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Attempting to add static directory");
- }
- this.webService.addStaticResources("/static", "/static");
-
- webService.start();
+ this.addWebServerHandlers(webService, metricsServlet, this.config);
+ this.webService.start();
// Refresh addresses, since the port might have been dynamically assigned
this.webServiceAddress = webAddress(config);
@@ -839,6 +799,107 @@ public class PulsarService implements AutoCloseable {
}
}
+ private void addWebServerHandlers(WebService webService,
+ PrometheusMetricsServlet metricsServlet,
+ ServiceConfiguration config)
+ throws PulsarServerException, PulsarClientException, MalformedURLException, ServletException,
+ DeploymentException {
+ Map<String, Object> attributeMap = Maps.newHashMap();
+ attributeMap.put(WebService.ATTRIBUTE_PULSAR_NAME, this);
+
+ Map<String, Object> vipAttributeMap = Maps.newHashMap();
+ vipAttributeMap.put(VipStatus.ATTRIBUTE_STATUS_FILE_PATH, config.getStatusFilePath());
+ vipAttributeMap.put(VipStatus.ATTRIBUTE_IS_READY_PROBE, (Supplier<Boolean>) () -> {
+ // Ensure the VIP status is only visible when the broker is fully initialized
+ return state == State.Started;
+ });
+ // Add admin rest resources
+ webService.addRestResources("/",
+ VipStatus.class.getPackage().getName(), false, vipAttributeMap);
+ webService.addRestResources("/",
+ "org.apache.pulsar.broker.web", false, attributeMap);
+ webService.addRestResources("/admin",
+ "org.apache.pulsar.broker.admin.v1", true, attributeMap);
+ webService.addRestResources("/admin/v2",
+ "org.apache.pulsar.broker.admin.v2", true, attributeMap);
+ webService.addRestResources("/admin/v3",
+ "org.apache.pulsar.broker.admin.v3", true, attributeMap);
+ webService.addRestResources("/lookup",
+ "org.apache.pulsar.broker.lookup", true, attributeMap);
+
+ // Add metrics servlet
+ webService.addServlet("/metrics",
+ new ServletHolder(metricsServlet),
+ false, attributeMap);
+
+ // Add websocket service
+ addWebSocketServiceHandler(webService, attributeMap, config);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Attempting to add static directory");
+ }
+ // Add static resources
+ webService.addStaticResources("/static", "/static");
+
+ // Add broker additional servlets
+ addBrokerAdditionalServlets(webService, attributeMap, config);
+ }
+
+ private void addBrokerAdditionalServlets(WebService webService,
+ Map<String, Object> attributeMap,
+ ServiceConfiguration config) {
+ if (this.getBrokerAdditionalServlets() != null) {
+ Collection<AdditionalServletWithClassLoader> additionalServletCollection =
+ this.getBrokerAdditionalServlets().getServlets().values();
+ for (AdditionalServletWithClassLoader servletWithClassLoader : additionalServletCollection) {
+ servletWithClassLoader.loadConfig(config);
+ AdditionalServlet additionalServlet = servletWithClassLoader.getServlet();
+ if (additionalServlet instanceof AdditionalServletWithPulsarService) {
+ ((AdditionalServletWithPulsarService) additionalServlet).setPulsarService(this);
+ }
+ webService.addServlet(servletWithClassLoader.getBasePath(), servletWithClassLoader.getServletHolder(),
+ config.isAuthenticationEnabled(), attributeMap);
+ LOG.info("Broker add additional servlet basePath {} ", servletWithClassLoader.getBasePath());
+ }
+ }
+ }
+
+ private void addWebSocketServiceHandler(WebService webService,
+ Map<String, Object> attributeMap,
+ ServiceConfiguration config)
+ throws PulsarServerException, PulsarClientException, MalformedURLException, ServletException,
+ DeploymentException {
+ if (config.isWebSocketServiceEnabled()) {
+ // Use local broker address to avoid different IP address when using a VIP for service discovery
+ this.webSocketService = new WebSocketService(null, config);
+ this.webSocketService.start();
+
+ final WebSocketServlet producerWebSocketServlet = new WebSocketProducerServlet(webSocketService);
+ webService.addServlet(WebSocketProducerServlet.SERVLET_PATH,
+ new ServletHolder(producerWebSocketServlet), true, attributeMap);
+ webService.addServlet(WebSocketProducerServlet.SERVLET_PATH_V2,
+ new ServletHolder(producerWebSocketServlet), true, attributeMap);
+
+ final WebSocketServlet consumerWebSocketServlet = new WebSocketConsumerServlet(webSocketService);
+ webService.addServlet(WebSocketConsumerServlet.SERVLET_PATH,
+ new ServletHolder(consumerWebSocketServlet), true, attributeMap);
+ webService.addServlet(WebSocketConsumerServlet.SERVLET_PATH_V2,
+ new ServletHolder(consumerWebSocketServlet), true, attributeMap);
+
+ final WebSocketServlet readerWebSocketServlet = new WebSocketReaderServlet(webSocketService);
+ webService.addServlet(WebSocketReaderServlet.SERVLET_PATH,
+ new ServletHolder(readerWebSocketServlet), true, attributeMap);
+ webService.addServlet(WebSocketReaderServlet.SERVLET_PATH_V2,
+ new ServletHolder(readerWebSocketServlet), true, attributeMap);
+
+ final WebSocketServlet pingPongWebSocketServlet = new WebSocketPingPongServlet(webSocketService);
+ webService.addServlet(WebSocketPingPongServlet.SERVLET_PATH,
+ new ServletHolder(pingPongWebSocketServlet), true, attributeMap);
+ webService.addServlet(WebSocketPingPongServlet.SERVLET_PATH_V2,
+ new ServletHolder(pingPongWebSocketServlet), true, attributeMap);
+ }
+ }
+
private void handleDeleteCluster(Notification notification) {
if (notification.getPath().startsWith(ClusterResources.CLUSTERS_ROOT)
&& notification.getType() == NotificationType.Deleted) {
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithPulsarService.java
similarity index 70%
copy from pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java
copy to pulsar-broker/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithPulsarService.java
index 2807dc4..23037dd 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithPulsarService.java
@@ -16,8 +16,13 @@
* specific language governing permissions and limitations
* under the License.
*/
+package org.apache.pulsar.broker.web.plugin.servlet;
+
+import org.apache.pulsar.broker.PulsarService;
/**
- * Pulsar proxy servlet plugin.
+ * The additional servlet with pulsarService interface for support additional servlet to get pulsarService.
*/
-package org.apache.pulsar.broker.web.plugin.servlet;
\ No newline at end of file
+public interface AdditionalServletWithPulsarService extends AdditionalServlet {
+ void setPulsarService(PulsarService pulsarService);
+}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java
similarity index 90%
copy from pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java
copy to pulsar-broker/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java
index 2807dc4..7da276a 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java
@@ -16,8 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-
/**
- * Pulsar proxy servlet plugin.
+ * Pulsar broker servlet plugin.
*/
package org.apache.pulsar.broker.web.plugin.servlet;
\ No newline at end of file
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerAdditionalServletTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerAdditionalServletTest.java
new file mode 100644
index 0000000..4b4c8db
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerAdditionalServletTest.java
@@ -0,0 +1,200 @@
+/**
+ * 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;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import javax.servlet.Servlet;
+import javax.servlet.ServletConfig;
+import javax.servlet.ServletException;
+import javax.servlet.ServletOutputStream;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import lombok.extern.slf4j.Slf4j;
+import okhttp3.OkHttpClient;
+import okhttp3.Response;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlet;
+import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServletWithClassLoader;
+import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServletWithPulsarService;
+import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlets;
+import org.apache.pulsar.common.configuration.PulsarConfiguration;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.servlet.ServletHolder;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class BrokerAdditionalServletTest extends MockedPulsarServiceBaseTest {
+
+ private final String BASE_PATH = "/additional/servlet";
+ private final String WITH_PULSAR_SERVICE_BASE_PATH = "/additional/servlet/with/pulsar/service";
+ private final String QUERY_PARAM = "param";
+
+ @Override
+ @BeforeClass
+ protected void setup() throws Exception {
+ internalSetup();
+ }
+
+ @Override
+ @AfterClass(alwaysRun = true)
+ protected void cleanup() throws Exception {
+ internalCleanup();
+ }
+
+ @Override
+ protected void beforePulsarStartMocks(PulsarService pulsar) throws Exception {
+ mockAdditionalServlet(pulsar);
+ }
+
+ private void mockAdditionalServlet(PulsarService pulsar) {
+ Servlet servlet = new OrdinaryServlet();
+
+ AdditionalServlet brokerAdditionalServlet = Mockito.mock(AdditionalServlet.class);
+ Mockito.when(brokerAdditionalServlet.getBasePath()).thenReturn(BASE_PATH);
+ Mockito.when(brokerAdditionalServlet.getServletHolder()).thenReturn(new ServletHolder(servlet));
+
+ AdditionalServletWithPulsarService brokerAdditionalServletWithPulsarService =
+ new AdditionalServletWithPulsarService() {
+ private PulsarService pulsarService;
+ @Override
+ public void setPulsarService(PulsarService pulsarService) {
+ this.pulsarService = pulsarService;
+ }
+
+ @Override
+ public void loadConfig(PulsarConfiguration pulsarConfiguration) {
+ // No-op
+ }
+
+ @Override
+ public String getBasePath() {
+ return WITH_PULSAR_SERVICE_BASE_PATH;
+ }
+
+ @Override
+ public ServletHolder getServletHolder() {
+ return new ServletHolder(new WithPulsarServiceServlet(pulsarService));
+ }
+
+ @Override
+ public void close() {
+ // No-op
+ }
+ };
+
+
+ AdditionalServlets brokerAdditionalServlets = Mockito.mock(AdditionalServlets.class);
+ Map<String, AdditionalServletWithClassLoader> map = new HashMap<>();
+ map.put("broker-additional-servlet", new AdditionalServletWithClassLoader(brokerAdditionalServlet, null));
+ map.put("broker-additional-servlet-with-pulsar-service", new AdditionalServletWithClassLoader(brokerAdditionalServletWithPulsarService, null));
+ Mockito.when(brokerAdditionalServlets.getServlets()).thenReturn(map);
+
+ Mockito.when(pulsar.getBrokerAdditionalServlets()).thenReturn(brokerAdditionalServlets);
+ }
+
+ @Test
+ public void test() throws IOException {
+ int httpPort = pulsar.getWebService().getListenPortHTTP().get();
+ log.info("pulsar webService httpPort {}", httpPort);
+ String paramValue = "value - " + RandomUtils.nextInt();
+ String response = httpGet("http://localhost:" + httpPort + BASE_PATH + "?" + QUERY_PARAM + "=" + paramValue);
+ Assert.assertEquals(response, paramValue);
+
+ String WithPulsarServiceParamValue = PulsarService.class.getName();
+ String withPulsarServiceResponse = httpGet("http://localhost:" + httpPort + WITH_PULSAR_SERVICE_BASE_PATH);
+ Assert.assertEquals(WithPulsarServiceParamValue, withPulsarServiceResponse);
+ }
+
+
+ private class OrdinaryServlet implements Servlet {
+ @Override
+ public void init(ServletConfig servletConfig) throws ServletException {
+ log.info("[init]");
+ }
+
+ @Override
+ public ServletConfig getServletConfig() {
+ log.info("[getServletConfig]");
+ return null;
+ }
+
+ @Override
+ public void service(ServletRequest servletRequest, ServletResponse servletResponse) throws ServletException,
+ IOException {
+ log.info("[service] path: {}", ((Request) servletRequest).getOriginalURI());
+ String value = servletRequest.getParameterMap().get(QUERY_PARAM)[0];
+ ServletOutputStream servletOutputStream = servletResponse.getOutputStream();
+ servletResponse.setContentLength(value.getBytes().length);
+ servletOutputStream.write(value.getBytes());
+ servletOutputStream.flush();
+ }
+
+ @Override
+ public String getServletInfo() {
+ log.info("[getServletInfo]");
+ return null;
+ }
+
+ @Override
+ public void destroy() {
+ log.info("[destroy]");
+ }
+ }
+
+
+ private class WithPulsarServiceServlet extends OrdinaryServlet {
+ private final PulsarService pulsarService;
+
+ public WithPulsarServiceServlet(PulsarService pulsar) {
+ this.pulsarService = pulsar;
+ }
+
+ @Override
+ public void service(ServletRequest servletRequest, ServletResponse servletResponse) throws ServletException,
+ IOException {
+ log.info("[service] path: {}", ((Request) servletRequest).getOriginalURI());
+ String value = pulsarService == null ? "null" : PulsarService.class.getName();
+ ServletOutputStream servletOutputStream = servletResponse.getOutputStream();
+ servletResponse.setContentLength(value.getBytes().length);
+ servletOutputStream.write(value.getBytes());
+ servletOutputStream.flush();
+ }
+ }
+
+ String httpGet(String url) throws IOException {
+ OkHttpClient client = new OkHttpClient();
+ okhttp3.Request request = new okhttp3.Request.Builder()
+ .get()
+ .url(url)
+ .build();
+
+ try (Response response = client.newCall(request).execute()) {
+ return response.body().string();
+ }
+ }
+
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
index 2dc99b9..f14c6e1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
@@ -238,6 +238,10 @@ public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport {
protected abstract void cleanup() throws Exception;
+ protected void beforePulsarStartMocks(PulsarService pulsar) throws Exception {
+ // No-op
+ }
+
protected void restartBroker() throws Exception {
stopBroker();
startBroker();
@@ -284,6 +288,7 @@ public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport {
conf.setBrokerShutdownTimeoutMs(0L);
PulsarService pulsar = spy(new PulsarService(conf));
setupBrokerMocks(pulsar);
+ beforePulsarStartMocks(pulsar);
pulsar.start();
log.info("Pulsar started. brokerServiceUrl: {} webServiceAddress: {}", pulsar.getBrokerServiceUrl(),
pulsar.getWebServiceAddress());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithPulsarServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithPulsarServiceTest.java
new file mode 100644
index 0000000..479c943
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithPulsarServiceTest.java
@@ -0,0 +1,85 @@
+/**
+ * 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.plugin.servlet;
+
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.AssertJUnit.assertSame;
+import static org.testng.AssertJUnit.assertTrue;
+import java.io.File;
+import java.nio.file.Paths;
+import java.util.Set;
+import org.apache.pulsar.common.nar.NarClassLoader;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.testng.IObjectFactory;
+import org.testng.annotations.ObjectFactory;
+import org.testng.annotations.Test;
+
+@PrepareForTest({
+ NarClassLoader.class
+})
+@PowerMockIgnore({"org.apache.logging.log4j.*"})
+public class AdditionalServletWithPulsarServiceTest {
+
+ // Necessary to make PowerMockito.mockStatic work with TestNG.
+ @ObjectFactory
+ public IObjectFactory getObjectFactory() {
+ return new org.powermock.modules.testng.PowerMockObjectFactory();
+ }
+
+ @Test
+ public void testLoadAdditionalServlet() throws Exception {
+ AdditionalServletDefinition def = new AdditionalServletDefinition();
+ def.setAdditionalServletClass(MockAdditionalServletWithClassLoader.class.getName());
+ def.setDescription("test-additional-servlet");
+
+ String archivePath = "/path/to/additional/servlet/nar";
+
+ AdditionalServletMetadata metadata = new AdditionalServletMetadata();
+ metadata.setDefinition(def);
+ metadata.setArchivePath(Paths.get(archivePath));
+
+ NarClassLoader mockLoader = mock(NarClassLoader.class);
+ when(mockLoader.getServiceDefinition(eq(AdditionalServletUtils.ADDITIONAL_SERVLET_FILE)))
+ .thenReturn(ObjectMapperFactory.getThreadLocalYaml().writeValueAsString(def));
+ Class additionalServletClass = MockAdditionalServletWithClassLoader.class;
+ when(mockLoader.loadClass(eq(MockAdditionalServletWithClassLoader.class.getName())))
+ .thenReturn(additionalServletClass);
+
+ PowerMockito.mockStatic(NarClassLoader.class);
+ PowerMockito.when(NarClassLoader.getFromArchive(
+ any(File.class),
+ any(Set.class),
+ any(ClassLoader.class),
+ any(String.class)
+ )).thenReturn(mockLoader);
+
+ AdditionalServletWithClassLoader returnedASWithCL = AdditionalServletUtils.load(metadata, "");
+ AdditionalServlet returnedPh = returnedASWithCL.getServlet();
+
+ assertSame(mockLoader, returnedASWithCL.getClassLoader());
+ assertTrue(returnedPh instanceof MockAdditionalServletWithClassLoader);
+ }
+}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/MockAdditionalServletWithClassLoader.java
similarity index 51%
copy from pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java
copy to pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/MockAdditionalServletWithClassLoader.java
index 2807dc4..6ff7b85 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/package-info.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/MockAdditionalServletWithClassLoader.java
@@ -16,8 +16,35 @@
* specific language governing permissions and limitations
* under the License.
*/
+package org.apache.pulsar.broker.web.plugin.servlet;
-/**
- * Pulsar proxy servlet plugin.
- */
-package org.apache.pulsar.broker.web.plugin.servlet;
\ No newline at end of file
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.common.configuration.PulsarConfiguration;
+import org.eclipse.jetty.servlet.ServletHolder;
+
+public class MockAdditionalServletWithClassLoader implements AdditionalServletWithPulsarService{
+ @Override
+ public void loadConfig(PulsarConfiguration pulsarConfiguration) {
+ // No-op
+ }
+
+ @Override
+ public String getBasePath() {
+ return null;
+ }
+
+ @Override
+ public ServletHolder getServletHolder() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+ // No-op
+ }
+
+ @Override
+ public void setPulsarService(PulsarService pulsarService) {
+ // No-op
+ }
+}