diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml index 1c60359525a96..fd11162d8ea3b 100644 --- a/pulsar-broker/pom.xml +++ b/pulsar-broker/pom.xml @@ -207,6 +207,20 @@ jersey-media-json-jackson + + org.glassfish.jersey.test-framework + jersey-test-framework-core + test + ${jersey.version} + + + + org.glassfish.jersey.test-framework.providers + jersey-test-framework-provider-grizzly2 + test + ${jersey.version} + + jakarta.activation jakarta.activation-api diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/v1/TopicLookup.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/v1/TopicLookup.java index 2b0b300634bfd..a2a2659a0fe4d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/v1/TopicLookup.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/v1/TopicLookup.java @@ -23,6 +23,7 @@ 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; @@ -30,6 +31,7 @@ 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); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/v2/TopicLookup.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/v2/TopicLookup.java index 031fe486d8997..0ac74cf782b90 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/v2/TopicLookup.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/v2/TopicLookup.java @@ -23,6 +23,7 @@ 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; @@ -30,12 +31,15 @@ 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); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/lookup/http/HttpTopicLookupv2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/lookup/http/HttpTopicLookupv2Test.java index db2b16634f33e..b65b084ccdc35 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/lookup/http/HttpTopicLookupv2Test.java +++ b/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 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 arg = ArgumentCaptor.forClass(Throwable.class); verify(asyncResponse1).resume(arg.capture()); @@ -182,7 +182,7 @@ 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 arg = ArgumentCaptor.forClass(Throwable.class); verify(asyncResponse).resume(arg.capture()); @@ -190,7 +190,7 @@ public void testValidateReplicationSettingsOnNamespace() throws Exception { 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 arg2 = ArgumentCaptor.forClass(Throwable.class); verify(asyncResponse2).resume(arg2.capture()); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/lookup/http/v2/TopicLookupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/lookup/http/v2/TopicLookupTest.java new file mode 100644 index 0000000000000..317b320a9a4d4 --- /dev/null +++ b/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()); + } + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/PulsarWebResourceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/PulsarWebResourceTest.java new file mode 100644 index 0000000000000..18e3e9613708c --- /dev/null +++ b/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(); + } +}