Prevent redirection of lookup requests from looping (#7200)

Author: Masahiro Sakamoto

pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/LookupResult.java

@@ -34,28 +34,35 @@ public enum Type {
 
     private final Type type;
     private final LookupData lookupData;
+    private final boolean authoritativeRedirect;
 
     public LookupResult(NamespaceEphemeralData namespaceEphemeralData) {
         this.type = Type.BrokerUrl;
+        this.authoritativeRedirect = false;
         this.lookupData = new LookupData(namespaceEphemeralData.getNativeUrl(),
                 namespaceEphemeralData.getNativeUrlTls(), namespaceEphemeralData.getHttpUrl(),
                 namespaceEphemeralData.getHttpUrlTls());
     }
 
-    public LookupResult(String httpUrl, String httpUrlTls, String brokerServiceUrl, String brokerServiceUrlTls) {
+    public LookupResult(String httpUrl, String httpUrlTls, String brokerServiceUrl, String brokerServiceUrlTls,
+            boolean authoritativeRedirect) {
         this.type = Type.RedirectUrl; // type = redirect => as current broker is
                                       // not owner and prepares LookupResult
                                       // with other broker's urls
+        this.authoritativeRedirect = authoritativeRedirect;
         this.lookupData = new LookupData(brokerServiceUrl, brokerServiceUrlTls, httpUrl, httpUrlTls);
     }
 
-    public LookupResult(String httpUrl, String httpUrlTls, String nativeUrl, String nativeUrlTls, Type type) {
+    public LookupResult(String httpUrl, String httpUrlTls, String nativeUrl, String nativeUrlTls, Type type,
+            boolean authoritativeRedirect) {
         this.type = type;
+        this.authoritativeRedirect = authoritativeRedirect;
         this.lookupData = new LookupData(nativeUrl, nativeUrlTls, httpUrl, httpUrlTls);
     }
 
     public LookupResult(NamespaceEphemeralData namespaceEphemeralData, String nativeUrl, String nativeUrlTls) {
         this.type = Type.BrokerUrl;
+        this.authoritativeRedirect = false;
         this.lookupData = new LookupData(nativeUrl, nativeUrlTls, namespaceEphemeralData.getHttpUrl(),
                 namespaceEphemeralData.getHttpUrlTls());
     }
@@ -68,6 +75,10 @@ public boolean isRedirect() {
         return type == Type.RedirectUrl;
     }
 
+    public boolean isAuthoritativeRedirect() {
+        return authoritativeRedirect;
+    }
+
     public LookupData getLookupData() {
         return lookupData;
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java

@@ -92,7 +92,7 @@ protected void internalLookupTopicAsync(TopicName topicName, boolean authoritati
             LookupResult result = optionalResult.get();
             // We have found either a broker that owns the topic, or a broker to which we should redirect the client to
             if (result.isRedirect()) {
-                boolean newAuthoritative = this.isLeaderBroker();
+                boolean newAuthoritative = result.isAuthoritativeRedirect();
                 URI redirect;
                 try {
                     String redirectUrl = isRequestHttps() ? result.getLookupData().getHttpUrlTls()
@@ -266,7 +266,7 @@ public static CompletableFuture<ByteBuf> lookupTopicAsync(PulsarService pulsarSe
 
                             LookupData lookupData = lookupResult.get().getLookupData();
                             if (lookupResult.get().isRedirect()) {
-                                boolean newAuthoritative = isLeaderBroker(pulsarService);
+                                boolean newAuthoritative = lookupResult.get().isAuthoritativeRedirect();
                                 lookupfuture.complete(
                                         newLookupResponse(lookupData.getBrokerUrl(), lookupData.getBrokerUrlTls(),
                                                 newAuthoritative, LookupType.Redirect, requestId, false));

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

@@ -426,6 +426,8 @@ private void searchForCandidateBroker(NamespaceBundle bundle,
                                           CompletableFuture<Optional<LookupResult>> lookupFuture, boolean authoritative,
                                           final String advertisedListenerName) {
         String candidateBroker = null;
+        boolean authoritativeRedirect = pulsar.getLeaderElectionService().isLeader();
+
         try {
             // check if this is Heartbeat or SLAMonitor namespace
             candidateBroker = checkHeartbeatNamespace(bundle);
@@ -438,7 +440,10 @@ private void searchForCandidateBroker(NamespaceBundle bundle,
             }
 
             if (candidateBroker == null) {
-                if (!this.loadManager.get().isCentralized()
+                if (authoritative) {
+                    // leader broker already assigned the current broker as owner
+                    candidateBroker = pulsar.getSafeWebServiceAddress();
+                } else if (!this.loadManager.get().isCentralized()
                         || pulsar.getLeaderElectionService().isLeader()
 
                         // If leader is not active, fallback to pick the least loaded from current broker loadmanager
@@ -450,14 +455,10 @@ private void searchForCandidateBroker(NamespaceBundle bundle,
                         return;
                     }
                     candidateBroker = availableBroker.get();
+                    authoritativeRedirect = true;
                 } else {
-                    if (authoritative) {
-                        // leader broker already assigned the current broker as owner
-                        candidateBroker = pulsar.getSafeWebServiceAddress();
-                    } else {
-                        // forward to leader broker to make assignment
-                        candidateBroker = pulsar.getLeaderElectionService().getCurrentLeader().getServiceUrl();
-                    }
+                    // forward to leader broker to make assignment
+                    candidateBroker = pulsar.getLeaderElectionService().getCurrentLeader().getServiceUrl();
                 }
             }
         } catch (Exception e) {
@@ -519,7 +520,7 @@ private void searchForCandidateBroker(NamespaceBundle bundle,
                 }
 
                 // Now setting the redirect url
-                createLookupResult(candidateBroker)
+                createLookupResult(candidateBroker, authoritativeRedirect)
                         .thenAccept(lookupResult -> lookupFuture.complete(Optional.of(lookupResult)))
                         .exceptionally(ex -> {
                             lookupFuture.completeExceptionally(ex);
@@ -533,7 +534,8 @@ private void searchForCandidateBroker(NamespaceBundle bundle,
         }
     }
 
-    protected CompletableFuture<LookupResult> createLookupResult(String candidateBroker) throws Exception {
+    protected CompletableFuture<LookupResult> createLookupResult(String candidateBroker, boolean authoritativeRedirect)
+            throws Exception {
 
         CompletableFuture<LookupResult> lookupFuture = new CompletableFuture<>();
         try {
@@ -546,7 +548,7 @@ protected CompletableFuture<LookupResult> createLookupResult(String candidateBro
                     ServiceLookupData lookupData = reportData.get();
                     lookupFuture.complete(new LookupResult(lookupData.getWebServiceUrl(),
                             lookupData.getWebServiceUrlTls(), lookupData.getPulsarServiceUrl(),
-                            lookupData.getPulsarServiceUrlTls()));
+                            lookupData.getPulsarServiceUrlTls(), authoritativeRedirect));
                 } else {
                     lookupFuture.completeExceptionally(new KeeperException.NoNodeException(path));
                 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java

@@ -358,13 +358,13 @@ public void testLoadReportDeserialize() throws Exception {
         ZkUtils.createFullPathOptimistic(pulsar.getZkClient(), path2,
                 ObjectMapperFactory.getThreadLocal().writeValueAsBytes(ld), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                 CreateMode.EPHEMERAL);
-        LookupResult result1 = pulsar.getNamespaceService().createLookupResult(candidateBroker1).get();
+        LookupResult result1 = pulsar.getNamespaceService().createLookupResult(candidateBroker1, false).get();
 
         // update to new load manager
         LoadManager oldLoadManager = pulsar.getLoadManager()
                 .getAndSet(new ModularLoadManagerWrapper(new ModularLoadManagerImpl()));
         oldLoadManager.stop();
-        LookupResult result2 = pulsar.getNamespaceService().createLookupResult(candidateBroker2).get();
+        LookupResult result2 = pulsar.getNamespaceService().createLookupResult(candidateBroker2, false).get();
         Assert.assertEquals(result1.getLookupData().getBrokerUrl(), candidateBroker1);
         Assert.assertEquals(result2.getLookupData().getBrokerUrl(), candidateBroker2);
         System.out.println(result2);