Support multiple advertisedListener for HTTP protocol

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/validator/MultipleListenerValidator.java

@@ -76,7 +76,8 @@ public static Map<String, AdvertisedListener> validateAndAnalysisAdvertisedListe
         final Map<String, AdvertisedListener> result = new LinkedHashMap<>();
         final Map<String, Set<String>> reverseMappings = new LinkedHashMap<>();
         for (final Map.Entry<String, List<String>> entry : listeners.entrySet()) {
-            if (entry.getValue().size() > 2) {
+            // one listener could configure at most 4 address: pulsar, pulsar+ssl, http, https
+            if (entry.getValue().size() > 4) {
                 throw new IllegalArgumentException("there are redundant configure for listener `" + entry.getKey()
                         + "`");
             }

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

@@ -428,9 +428,21 @@ private CompletableFuture<Optional<LookupResult>> findBrokerServiceUrl(
                         } else {
                             URI url = listener.getBrokerServiceUrl();
                             URI urlTls = listener.getBrokerServiceUrlTls();
-                            future.complete(Optional.of(new LookupResult(nsData.get(),
-                                    url == null ? null : url.toString(),
-                                    urlTls == null ? null : urlTls.toString())));
+                            URI http = listener.getBrokerHttpUrl();
+                            URI https = listener.getBrokerHttpsUrl();
+                            if (http == null && https == null) {
+                                future.complete(Optional.of(
+                                        new LookupResult(nsData.get(),
+                                                url == null ? null : url.toString(),
+                                                urlTls == null ? null : urlTls.toString())));
+                            } else {
+                                future.complete(Optional.of(
+                                        new LookupResult(http == null ? null : http.toString(),
+                                                https == null ? null : https.toString(),
+                                                url == null ? null : url.toString(),
+                                                urlTls == null ? null : urlTls.toString(),
+                                                LookupResult.Type.BrokerUrl, false)));
+                            }
                         }
                         return;
                     } else {
@@ -576,11 +588,21 @@ private void searchForCandidateBroker(NamespaceBundle bundle,
                             } else {
                                 URI url = listener.getBrokerServiceUrl();
                                 URI urlTls = listener.getBrokerServiceUrlTls();
-                                lookupFuture.complete(Optional.of(
-                                        new LookupResult(ownerInfo,
-                                                url == null ? null : url.toString(),
-                                                urlTls == null ? null : urlTls.toString())));
-                                return;
+                                URI http = listener.getBrokerHttpUrl();
+                                URI https = listener.getBrokerHttpsUrl();
+                                if (http == null && https == null) {
+                                    lookupFuture.complete(Optional.of(
+                                            new LookupResult(ownerInfo,
+                                                    url == null ? null : url.toString(),
+                                                    urlTls == null ? null : urlTls.toString())));
+                                } else {
+                                    lookupFuture.complete(Optional.of(
+                                            new LookupResult(http == null ? null : http.toString(),
+                                                    https == null ? null : https.toString(),
+                                                    url == null ? null : url.toString(),
+                                                    urlTls == null ? null : urlTls.toString(),
+                                                    LookupResult.Type.BrokerUrl, false)));
+                                }
                             }
                         } else {
                             lookupFuture.complete(Optional.of(new LookupResult(ownerInfo)));

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

@@ -99,6 +99,8 @@ public abstract class PulsarWebResource {
 
     static final String ORIGINAL_PRINCIPAL_HEADER = "X-Original-Principal";
 
+    static final String LISTENER_HEADER = "X-Pulsar-ListenerName";
+
     @Context
     protected ServletContext servletContext;
 
@@ -592,10 +594,12 @@ protected CompletableFuture<Boolean> isBundleOwnedByAnyBroker(NamespaceName fqnn
         NamespaceBundle nsBundle = validateNamespaceBundleRange(fqnn, bundles, bundleRange);
         NamespaceService nsService = pulsar().getNamespaceService();
 
+        String advertisedListener = httpRequest == null ? null : httpRequest.getHeader(LISTENER_HEADER);
         LookupOptions options = LookupOptions.builder()
                 .authoritative(false)
                 .requestHttps(isRequestHttps())
                 .readOnly(true)
+                .advertisedListenerName(advertisedListener)
                 .loadTopicsInBundle(false).build();
         return nsService.getWebServiceUrlAsync(nsBundle, options).thenApply(optionUrl -> optionUrl.isPresent());
     }
@@ -642,10 +646,13 @@ public void validateBundleOwnership(NamespaceBundle bundle, boolean authoritativ
             // - If authoritative is false and this broker is not leader, forward to leader
             // - If authoritative is false and this broker is leader, determine owner and forward w/ authoritative=true
             // - If authoritative is true, own the namespace and continue
+            String advertisedListener = httpRequest == null ? null : httpRequest.getHeader(LISTENER_HEADER);
+
             LookupOptions options = LookupOptions.builder()
                     .authoritative(authoritative)
                     .requestHttps(isRequestHttps())
                     .readOnly(readOnly)
+                    .advertisedListenerName(advertisedListener)
                     .loadTopicsInBundle(false).build();
             Optional<URL> webUrl = nsService.getWebServiceUrl(bundle, options);
             // Ensure we get a url
@@ -735,11 +742,14 @@ protected void validateTopicOwnership(TopicName topicName, boolean authoritative
     protected CompletableFuture<Void> validateTopicOwnershipAsync(TopicName topicName, boolean authoritative) {
         NamespaceService nsService = pulsar().getNamespaceService();
 
+        String advertisedListener = httpRequest == null ? null : httpRequest.getHeader(LISTENER_HEADER);
+
         LookupOptions options = LookupOptions.builder()
                 .authoritative(authoritative)
                 .requestHttps(isRequestHttps())
                 .readOnly(false)
                 .loadTopicsInBundle(false)
+                .advertisedListenerName(advertisedListener)
                 .build();
 
         return nsService.getWebServiceUrlAsync(topicName, options)

pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerBaseTest.java

@@ -146,6 +146,23 @@ public final List<PulsarAdmin> getAllAdmins() {
         return Collections.unmodifiableList(admins);
     }
 
+    public final List<PulsarAdmin> getAllAdmins(String listener) throws Exception {
+        List<PulsarAdmin> admins = new ArrayList<>(numberOfAdditionalBrokers() + 1);
+
+        PulsarAdminBuilder pulsarAdminBuilder = PulsarAdmin.builder().serviceHttpUrl(
+            getPulsar().getWebServiceAddress() != null ? getPulsar().getWebServiceAddress()
+                    : getPulsar().getWebServiceAddressTls()).listenerName(listener);
+        admins.add(pulsarAdminBuilder.build());
+
+        for (PulsarService broker : additionalBrokers) {
+            pulsarAdminBuilder = PulsarAdmin.builder().serviceHttpUrl(broker.getWebServiceAddress() != null
+                    ? broker.getWebServiceAddress() : broker.getWebServiceAddressTls()).listenerName(listener);
+            admins.add(pulsarAdminBuilder.build());
+        }
+
+        return Collections.unmodifiableList(admins);
+    }
+
     public final List<PulsarClient> getAllClients() {
         List<PulsarClient> clients = new ArrayList<>(numberOfAdditionalBrokers() + 1);
         clients.add(pulsarClient);

pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/MultiHTTPAdvertisedListenersTest.java

@@ -0,0 +1,166 @@
+/**
+ * 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.loadbalance;
+
+import java.net.URI;
+import java.util.Optional;
+import lombok.Cleanup;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.util.PortManager;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpHeaders;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.util.EntityUtils;
+import org.apache.pulsar.broker.MultiBrokerBaseTest;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.lookup.data.LookupData;
+import org.apache.pulsar.common.policies.data.TopicStats;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+
+@Slf4j
+@Test(groups = "broker")
+public class MultiHTTPAdvertisedListenersTest extends MultiBrokerBaseTest {
+    @Override
+    protected int numberOfAdditionalBrokers() {
+        return 1;
+    }
+
+    @Override
+    protected void doInitConf() throws Exception {
+        super.doInitConf();
+
+        updateConfig(conf);
+    }
+
+    @Override
+    protected ServiceConfiguration createConfForAdditionalBroker(int additionalBrokerIndex) {
+        ServiceConfiguration conf = super.createConfForAdditionalBroker(additionalBrokerIndex);
+        updateConfig(conf);
+        return conf;
+    }
+
+    private void updateConfig(ServiceConfiguration conf) {
+        int pulsarPort = PortManager.nextFreePort();
+        int httpPort = PortManager.nextFreePort();
+        int httpsPort = PortManager.nextFreePort();
+
+        // Use invalid domain name as identifier and instead make sure the advertised listeners work as intended
+        conf.setAdvertisedListeners(
+                "internal:pulsar://localhost:" + pulsarPort +
+                ",internal:http://localhost:" + httpPort +
+                ",internal:https://localhost:" + httpsPort +
+                ",external:pulsar://externalip:" + pulsarPort +
+                ",external:http://externalip:" + httpPort +
+                ",external:https://externalip:" + httpsPort);
+        conf.setBrokerServicePort(Optional.of(pulsarPort));
+        conf.setWebServicePort(Optional.of(httpPort));
+        conf.setWebServicePortTls(Optional.of(httpsPort));
+    }
+
+    @Test
+    public void testLookupWithoutListener() throws Exception {
+        HttpGet request =
+                new HttpGet(pulsar.getWebServiceAddress() + "/lookup/v2/topic/persistent/public/default/my-topic");
+        request.addHeader(HttpHeaders.CONTENT_TYPE, "application/json");
+        request.addHeader(HttpHeaders.ACCEPT, "application/json");
+        final String topic = "my-topic";
+
+        @Cleanup
+        CloseableHttpClient httpClient = HttpClients.createDefault();
+
+        @Cleanup
+        CloseableHttpResponse response = httpClient.execute(request);
+
+        HttpEntity entity = response.getEntity();
+        LookupData ld = ObjectMapperFactory.getThreadLocal().readValue(EntityUtils.toString(entity), LookupData.class);
+        System.err.println("Lookup data: " + ld);
+
+        assertEquals(new URI(ld.getBrokerUrl()).getHost(), "localhost");
+        assertEquals(new URI(ld.getHttpUrl()).getHost(), "localhost");
+        assertEquals(new URI(ld.getHttpUrlTls()).getHost(), "localhost");
+
+
+        // Produce data
+        @Cleanup
+        Producer<String> p = pulsarClient.newProducer(Schema.STRING)
+                .topic(topic)
+                .create();
+
+        p.send("hello");
+
+        // Verify we can get the correct HTTP redirect to the advertised listener
+        for (PulsarAdmin a : getAllAdmins()) {
+            TopicStats s = a.topics().getStats(topic);
+            assertNotNull(a.lookups().lookupTopic(topic));
+            assertEquals(s.getPublishers().size(), 1);
+        }
+    }
+
+    @Test
+    public void testLookupWithListener() throws Exception {
+        HttpGet request =
+                new HttpGet(pulsar.getWebServiceAddress() + "/lookup/v2/topic/persistent/public/default/my-topic");
+        request.addHeader(HttpHeaders.CONTENT_TYPE, "application/json");
+        request.addHeader(HttpHeaders.ACCEPT, "application/json");
+        request.addHeader("X-Pulsar-ListenerName", "external");
+        final String topic = "my-topic";
+
+        @Cleanup
+        CloseableHttpClient httpClient = HttpClients.createDefault();
+
+        @Cleanup
+        CloseableHttpResponse response = httpClient.execute(request);
+
+        HttpEntity entity = response.getEntity();
+        LookupData ld = ObjectMapperFactory.getThreadLocal().readValue(EntityUtils.toString(entity), LookupData.class);
+        System.err.println("Lookup data: " + ld);
+
+        assertEquals(new URI(ld.getBrokerUrl()).getHost(), "externalip");
+        assertEquals(new URI(ld.getHttpUrl()).getHost(), "externalip");
+        assertEquals(new URI(ld.getHttpUrlTls()).getHost(), "externalip");
+
+
+        // Produce data
+        @Cleanup
+        Producer<String> p = pulsarClient.newProducer(Schema.STRING)
+                .topic(topic)
+                .create();
+
+        p.send("hello");
+
+        // Verify we can get the correct HTTP redirect to the advertised listener
+        for (PulsarAdmin a : getAllAdmins("internal")) {
+            TopicStats s = a.topics().getStats(topic);
+            assertNotNull(a.lookups().lookupTopic(topic));
+            assertEquals(s.getPublishers().size(), 1);
+            a.close();
+        }
+    }
+
+}

pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/PulsarAdminBuilder.java

@@ -327,4 +327,11 @@ PulsarAdminBuilder authentication(String authPluginClassName, Map<String, String
      */
     PulsarAdminBuilder setContextClassLoader(ClassLoader clientBuilderClassLoader);
 
+    /**
+     * Listener name for lookup.
+     *
+     * @param listenerName
+     */
+    PulsarAdminBuilder listenerName(String listenerName);
+
 }

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java

@@ -62,9 +62,18 @@ public abstract class BaseResource {
     protected final Authentication auth;
     protected final long readTimeoutMs;
 
+    protected final String advertisedListener;
+
+    static final String LISTENER_HEADER = "X-Pulsar-ListenerName";
+
     protected BaseResource(Authentication auth, long readTimeoutMs) {
+        this(auth, readTimeoutMs, null);
+    }
+
+    protected BaseResource(Authentication auth, long readTimeoutMs, String advertisedListener) {
         this.auth = auth;
         this.readTimeoutMs = readTimeoutMs;
+        this.advertisedListener = advertisedListener;
     }
 
     public Builder request(final WebTarget target) throws PulsarAdminException {
@@ -106,6 +115,10 @@ public CompletableFuture<Builder> requestAsync(final WebTarget target) {
                             headers.forEach(entry -> builder.header(entry.getKey(), entry.getValue()));
                         }
                     }
+
+                    if (advertisedListener != null) {
+                        builder.header(LISTENER_HEADER, advertisedListener);
+                    }
                     builderFuture.complete(builder);
                 } catch (Throwable t) {
                     builderFuture.completeExceptionally(new GettingAuthenticationDataException(t));

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BookiesImpl.java

@@ -33,7 +33,11 @@ public class BookiesImpl extends BaseResource implements Bookies {
     private final WebTarget adminBookies;
 
     public BookiesImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
-        super(auth, readTimeoutMs);
+        this(web, auth, readTimeoutMs, null);
+    }
+
+    public BookiesImpl(WebTarget web, Authentication auth, long readTimeoutMs, String advertisedListener) {
+        super(auth, readTimeoutMs, advertisedListener);
         adminBookies = web.path("/admin/v2/bookies");
     }
 

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokerStatsImpl.java

@@ -39,11 +39,16 @@ public class BrokerStatsImpl extends BaseResource implements BrokerStats {
     private final WebTarget adminV2BrokerStats;
 
     public BrokerStatsImpl(WebTarget target, Authentication auth, long readTimeoutMs) {
-        super(auth, readTimeoutMs);
+        this(target, auth, readTimeoutMs, null);
+    }
+
+    public BrokerStatsImpl(WebTarget target, Authentication auth, long readTimeoutMs, String advertisedListener) {
+        super(auth, readTimeoutMs, advertisedListener);
         adminBrokerStats = target.path("/admin/broker-stats");
         adminV2BrokerStats = target.path("/admin/v2/broker-stats");
     }
 
+
     @Override
     public String getMetrics() throws PulsarAdminException {
         return sync(this::getMetricsAsync);

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java

@@ -38,7 +38,11 @@ public class BrokersImpl extends BaseResource implements Brokers {
     private final WebTarget adminBrokers;
 
     public BrokersImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
-        super(auth, readTimeoutMs);
+        this(web, auth, readTimeoutMs, null);
+    }
+
+    public BrokersImpl(WebTarget web, Authentication auth, long readTimeoutMs, String advertisedListener) {
+        super(auth, readTimeoutMs, advertisedListener);
         adminBrokers = web.path("admin/v2/brokers");
     }