When running standalone broker, force the proxy redirection flag (apa…

…che#1449)
ivankelly · Mar 28, 2018 · 8f6e33d · 8f6e33d
1 parent ffd6f21
commit 8f6e33d
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 1 deletion.
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
@@ -59,6 +59,9 @@ public class ServiceConfiguration implements PulsarConfiguration {
     // Enable the WebSocket API service
     private boolean webSocketServiceEnabled = false;
 
+    // Flag to control features that are meant to be used when running in standalone mode
+    private boolean isRunningStandalone = false;
+
     // Name of the cluster to which this broker belongs to
     @FieldContext(required = true)
     private String clusterName;
@@ -1517,4 +1520,12 @@ public void setFunctionsWorkerEnabled(boolean enabled) {
     public boolean isFunctionsWorkerEnabled() {
         return functionsWorkerEnabled;
     }
+
+    public boolean isRunningStandalone() {
+        return isRunningStandalone;
+    }
+
+    public void setRunningStandalone(boolean isRunningStandalone) {
+        this.isRunningStandalone = isRunningStandalone;
+    }
 }
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandaloneStarter.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandaloneStarter.java
@@ -134,6 +134,7 @@ public PulsarStandaloneStarter(String[] args) throws Exception {
         // Set ZK server's host to localhost
         config.setZookeeperServers(zkServers + ":" + zkPort);
         config.setGlobalZookeeperServers(zkServers + ":" + zkPort);
+        config.setRunningStandalone(true);
 
         Runtime.getRuntime().addShutdownHook(new Thread() {
             public void run() {

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookup.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookup.java
@@ -288,9 +288,14 @@ public static CompletableFuture<ByteBuf> lookupTopicAsync(PulsarService pulsarSe
                                         newLookupResponse(lookupData.getBrokerUrl(), lookupData.getBrokerUrlTls(),
                                                 newAuthoritative, LookupType.Redirect, requestId, false));
                             } else {
+                                // When running in standalone mode we want to redirect the client through the service
+                                // url, so that the advertised address configuration is not relevant anymore.
+                                boolean redirectThroughServiceUrl = pulsarService.getConfiguration()
+                                        .isRunningStandalone();
+
                                 lookupfuture.complete(
                                         newLookupResponse(lookupData.getBrokerUrl(), lookupData.getBrokerUrlTls(),
-                                                true /* authoritative */, LookupType.Connect, requestId, false));
+                                                true /* authoritative */, LookupType.Connect, requestId, redirectThroughServiceUrl));
                             }
                         }).exceptionally(ex -> {
                             if (ex instanceof CompletionException && ex.getCause() instanceof IllegalStateException) {