[PIP 95][Issue 12040][web] Topic lookup with listener header

pulsar-broker/pom.xml

@@ -207,6 +207,20 @@
       <artifactId>jersey-media-json-jackson</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>org.glassfish.jersey.test-framework</groupId>
+      <artifactId>jersey-test-framework-core</artifactId>
+      <scope>test</scope>
+      <version>${jersey.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.glassfish.jersey.test-framework.providers</groupId>
+      <artifactId>jersey-test-framework-provider-grizzly2</artifactId>
+      <scope>test</scope>
+      <version>${jersey.version}</version>
+    </dependency>
+
     <dependency>
       <groupId>jakarta.activation</groupId>
       <artifactId>jakarta.activation-api</artifactId>

pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/v1/TopicLookup.java

@@ -23,13 +23,15 @@
 import javax.ws.rs.DefaultValue;
 import javax.ws.rs.Encoded;
 import javax.ws.rs.GET;
+import javax.ws.rs.HeaderParam;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
 import javax.ws.rs.QueryParam;
 import javax.ws.rs.container.AsyncResponse;
 import javax.ws.rs.container.Suspended;
 import javax.ws.rs.core.MediaType;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.broker.lookup.TopicLookupBase;
 import org.apache.pulsar.broker.web.NoSwaggerDocumentation;
 import org.apache.pulsar.common.naming.TopicName;
@@ -48,6 +50,8 @@
 @NoSwaggerDocumentation
 public class TopicLookup extends TopicLookupBase {
 
+    static final String LISTENERNAME_HEADER = "X-Pulsar-ListenerName";
+
     @GET
     @Path("{topic-domain}/{property}/{cluster}/{namespace}/{topic}")
     @Produces(MediaType.APPLICATION_JSON)
@@ -58,8 +62,12 @@ public void lookupTopicAsync(@PathParam("topic-domain") String topicDomain, @Pat
             @PathParam("topic") @Encoded String encodedTopic,
             @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
             @Suspended AsyncResponse asyncResponse,
-            @QueryParam("listenerName") String listenerName) {
+            @QueryParam("listenerName") String listenerName,
+            @HeaderParam(LISTENERNAME_HEADER) String listenerNameHeader) {
         TopicName topicName = getTopicName(topicDomain, property, cluster, namespace, encodedTopic);
+        if (StringUtils.isEmpty(listenerName) && StringUtils.isNotEmpty(listenerNameHeader)) {
+            listenerName = listenerNameHeader;
+        }
         internalLookupTopicAsync(topicName, authoritative, asyncResponse, listenerName);
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/v2/TopicLookup.java

@@ -23,19 +23,23 @@
 import javax.ws.rs.DefaultValue;
 import javax.ws.rs.Encoded;
 import javax.ws.rs.GET;
+import javax.ws.rs.HeaderParam;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
 import javax.ws.rs.QueryParam;
 import javax.ws.rs.container.AsyncResponse;
 import javax.ws.rs.container.Suspended;
 import javax.ws.rs.core.MediaType;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.broker.lookup.TopicLookupBase;
 import org.apache.pulsar.common.naming.TopicName;
 
 @Path("/v2/topic")
 public class TopicLookup extends TopicLookupBase {
 
+    static final String LISTENERNAME_HEADER = "X-Pulsar-ListenerName";
+
     @GET
     @Path("{topic-domain}/{tenant}/{namespace}/{topic}")
     @Produces(MediaType.APPLICATION_JSON)
@@ -45,8 +49,12 @@ public void lookupTopicAsync(@PathParam("topic-domain") String topicDomain, @Pat
             @PathParam("namespace") String namespace, @PathParam("topic") @Encoded String encodedTopic,
             @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
             @Suspended AsyncResponse asyncResponse,
-            @QueryParam("listenerName") String listenerName) {
+            @QueryParam("listenerName") String listenerName,
+            @HeaderParam(LISTENERNAME_HEADER) String listenerNameHeader) {
         TopicName topicName = getTopicName(topicDomain, tenant, namespace, encodedTopic);
+        if (StringUtils.isEmpty(listenerName) && StringUtils.isNotEmpty(listenerNameHeader)) {
+            listenerName = listenerNameHeader;
+        }
         internalLookupTopicAsync(topicName, authoritative, asyncResponse, listenerName);
     }
 

pulsar-broker/src/test/java/org/apache/pulsar/broker/lookup/http/HttpTopicLookupv2Test.java

@@ -116,7 +116,7 @@ public void crossColoLookup() throws Exception {
 
         AsyncResponse asyncResponse = mock(AsyncResponse.class);
         destLookup.lookupTopicAsync(TopicDomain.persistent.value(), "myprop", "usc", "ns2", "topic1", false,
-                asyncResponse, null);
+                asyncResponse, null, null);
 
         ArgumentCaptor<Throwable> arg = ArgumentCaptor.forClass(Throwable.class);
         verify(asyncResponse).resume(arg.capture());
@@ -146,7 +146,7 @@ public void testNotEnoughLookupPermits() throws Exception {
 
         AsyncResponse asyncResponse1 = mock(AsyncResponse.class);
         destLookup.lookupTopicAsync(TopicDomain.persistent.value(), "myprop", "usc", "ns2", "topic1", false,
-                asyncResponse1, null);
+                asyncResponse1, null, null);
 
         ArgumentCaptor<Throwable> arg = ArgumentCaptor.forClass(Throwable.class);
         verify(asyncResponse1).resume(arg.capture());
@@ -182,15 +182,15 @@ public void testValidateReplicationSettingsOnNamespace() throws Exception {
 
         AsyncResponse asyncResponse = mock(AsyncResponse.class);
         destLookup.lookupTopicAsync(TopicDomain.persistent.value(), property, cluster, ns1, "empty-cluster",
-                false, asyncResponse, null);
+                false, asyncResponse, null, null);
 
         ArgumentCaptor<Throwable> arg = ArgumentCaptor.forClass(Throwable.class);
         verify(asyncResponse).resume(arg.capture());
         assertEquals(arg.getValue().getClass(), RestException.class);
 
         AsyncResponse asyncResponse2 = mock(AsyncResponse.class);
         destLookup.lookupTopicAsync(TopicDomain.persistent.value(), property, cluster, ns2,
-                "invalid-localCluster", false, asyncResponse2, null);
+                "invalid-localCluster", false, asyncResponse2, null, null);
         ArgumentCaptor<Throwable> arg2 = ArgumentCaptor.forClass(Throwable.class);
         verify(asyncResponse2).resume(arg2.capture());
 

pulsar-broker/src/test/java/org/apache/pulsar/broker/lookup/http/v2/TopicLookupTest.java

@@ -0,0 +1,83 @@
+/**
+ * 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.lookup.http.v2;
+
+import javax.ws.rs.container.AsyncResponse;
+import javax.ws.rs.core.Response;
+import org.apache.pulsar.broker.lookup.v2.TopicLookup;
+import org.apache.pulsar.broker.web.PulsarWebResourceTest;
+import org.apache.pulsar.common.lookup.data.LookupData;
+import org.apache.pulsar.common.naming.TopicName;
+import org.glassfish.jersey.server.ResourceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.spy;
+import static org.testng.Assert.assertEquals;
+
+/**
+ * TopicLookup V2 API unit tests.
+ */
+@Test(groups = "broker")
+public class TopicLookupTest extends PulsarWebResourceTest {
+
+    private static final String TOPIC_PATH = "/v2/topic/persistent/public/testns/testtopic";
+
+    private TestableTopicLookup resource;
+
+    @Override
+    protected ResourceConfig configure() {
+        resource = spy(new TestableTopicLookup());
+        return new ResourceConfig().register(resource);
+    }
+
+    @Test
+    public void testListenerName() {
+        Response response;
+        // verify query param
+        response = target(TOPIC_PATH).queryParam("listenerName", "query").request().get();
+        assertEquals(response.getStatus(), 200);
+        assertEquals(resource.actualListenerName, "query");
+
+        // verify header param
+        response = target(TOPIC_PATH).request().header("X-Pulsar-ListenerName", "header").get();
+        assertEquals(response.getStatus(), 200);
+        assertEquals(resource.actualListenerName, "header");
+
+        // verify that query param supersedes the header param
+        response = target(TOPIC_PATH).queryParam("listenerName", "query")
+                .request().header("X-Pulsar-ListenerName", "header").get();
+        assertEquals(response.getStatus(), 200);
+        assertEquals(resource.actualListenerName, "query");
+    }
+
+    private static class TestableTopicLookup extends TopicLookup {
+        private String actualListenerName;
+
+        @Override
+        protected void internalLookupTopicAsync(TopicName topicName, boolean authoritative, AsyncResponse asyncResponse,
+                String listenerName) {
+            this.actualListenerName = listenerName;
+            asyncResponse.resume(new LookupData());
+        }
+    }
+}

pulsar-broker/src/test/java/org/apache/pulsar/broker/web/PulsarWebResourceTest.java

@@ -0,0 +1,104 @@
+/**
+ * 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 javax.servlet.ServletContext;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.Feature;
+import javax.ws.rs.core.FeatureContext;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.glassfish.jersey.server.ResourceConfig;
+import org.glassfish.jersey.servlet.ServletContainer;
+import org.glassfish.jersey.test.DeploymentContext;
+import org.glassfish.jersey.test.JerseyTestNg;
+import org.glassfish.jersey.test.ServletDeploymentContext;
+import org.glassfish.jersey.test.TestProperties;
+import org.glassfish.jersey.test.grizzly.GrizzlyWebTestContainerFactory;
+import org.glassfish.jersey.test.spi.TestContainerException;
+import org.glassfish.jersey.test.spi.TestContainerFactory;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+/**
+ * A base class for testing subclasses of {@link PulsarWebResource}.
+ */
+public abstract class PulsarWebResourceTest extends JerseyTestNg.ContainerPerClassTest {
+
+    protected ServiceConfiguration config;
+    protected PulsarService pulsar;
+
+    protected PulsarWebResourceTest() {
+        config = new ServiceConfiguration();
+
+        pulsar = mock(PulsarService.class);
+        doReturn(config).when(pulsar).getConfig();
+        doReturn(config).when(pulsar).getConfiguration();
+
+        set(TestProperties.CONTAINER_PORT, 0);
+    }
+
+    @BeforeClass(alwaysRun = true)
+    @Override
+    public void setUp() throws Exception {
+        super.setUp();
+    }
+
+    @AfterClass(alwaysRun = true)
+    @Override
+    public void tearDown() throws Exception {
+        super.tearDown();
+    }
+
+    /**
+     * Creates a JAX-RS resource configuration for test purposes.
+     */
+    @Override
+    protected abstract ResourceConfig configure();
+
+    /**
+     * Creates a test container factory with servlet support.
+     */
+    @Override
+    protected TestContainerFactory getTestContainerFactory() throws TestContainerException {
+        return new GrizzlyWebTestContainerFactory();
+    }
+
+    /**
+     * Configures a deployment context for JAX-RS.
+     */
+    @Override
+    protected DeploymentContext configureDeployment() {
+        ResourceConfig app = configure();
+        app.register(new Feature() {
+            @Context
+            ServletContext servletContext;
+
+            @Override
+            public boolean configure(FeatureContext context) {
+                servletContext.setAttribute(WebService.ATTRIBUTE_PULSAR_NAME, pulsar);
+                return true;
+            }
+        });
+        return ServletDeploymentContext.forServlet(new ServletContainer(app)).build();
+    }
+}