[fix][client] fix producer/consumer perform lookup for migrated topic (…

…apache#21356)

Co-authored-by: Rajan Dhabalia <rdhabalia@oath.com>
(cherry picked from commit d09642c)
(cherry picked from commit d4f3c59)

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java

@@ -96,9 +96,16 @@ protected void grabCnx() {
         try {
             CompletableFuture<ClientCnx> cnxFuture;
             if (state.redirectedClusterURI != null) {
-                InetSocketAddress address = InetSocketAddress.createUnresolved(state.redirectedClusterURI.getHost(),
-                        state.redirectedClusterURI.getPort());
-                cnxFuture = state.client.getConnection(address, address, randomKeyForSelectConnection);
+                if (state.topic == null) {
+                    InetSocketAddress address = InetSocketAddress.createUnresolved(state.redirectedClusterURI.getHost(),
+                            state.redirectedClusterURI.getPort());
+                    cnxFuture = state.client.getConnection(address, address, randomKeyForSelectConnection);
+                } else {
+                    // once, client receives redirection url, client has to perform lookup on migrated
+                    // cluster to find the broker that owns the topic and then create connection.
+                    // below method, performs the lookup for a given topic and then creates connection
+                    cnxFuture = state.client.getConnection(state.topic, (state.redirectedClusterURI.toString()));
+                }
             } else if (state.topic == null) {
                 cnxFuture = state.client.getConnectionToServiceUrl();
             } else {

pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java

@@ -124,5 +124,4 @@ CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadata(TopicNam
      */
     CompletableFuture<GetTopicsResult> getTopicsUnderNamespace(NamespaceName namespace, Mode mode,
                                                                String topicPattern, String topicsHash);
-
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java

@@ -33,6 +33,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -106,6 +107,7 @@ public class PulsarClientImpl implements PulsarClient {
 
     private final boolean createdScheduledProviders;
     private LookupService lookup;
+    private Map<String, LookupService> urlLookupMap = new ConcurrentHashMap<>();
     private final ConnectionPool cnxPool;
     @Getter
     private final Timer timer;
@@ -962,6 +964,23 @@ public CompletableFuture<ClientCnx> getConnection(final String topic) {
                 .thenCompose(pair -> getConnection(pair.getLeft(), pair.getRight(), cnxPool.genRandomKeyToSelectCon()));
     }
 
+    public CompletableFuture<ClientCnx> getConnection(final String topic, final String url) {
+        TopicName topicName = TopicName.get(topic);
+        return getLookup(url).getBroker(topicName)
+                .thenCompose(pair -> getConnection(pair.getLeft(), pair.getRight(), cnxPool.genRandomKeyToSelectCon()));
+    }
+
+    public LookupService getLookup(String serviceUrl) {
+        return urlLookupMap.computeIfAbsent(serviceUrl, url -> {
+            try {
+                return createLookup(serviceUrl);
+            } catch (PulsarClientException e) {
+                log.warn("Failed to update url to lookup service {}, {}", url, e.getMessage());
+                throw new IllegalStateException("Failed to update url " + url);
+            }
+        });
+    }
+
     public CompletableFuture<ClientCnx> getConnectionToServiceUrl() {
         if (!(lookup instanceof BinaryProtoLookupService)) {
             return FutureUtil.failedFuture(new PulsarClientException.InvalidServiceURL(
@@ -1020,10 +1039,14 @@ public LookupService getLookup() {
     }
 
     public void reloadLookUp() throws PulsarClientException {
-        if (conf.getServiceUrl().startsWith("http")) {
-            lookup = new HttpLookupService(conf, eventLoopGroup);
+        lookup = createLookup(conf.getServiceUrl());
+    }
+
+    public LookupService createLookup(String url) throws PulsarClientException {
+        if (url.startsWith("http")) {
+            return new HttpLookupService(conf, eventLoopGroup);
         } else {
-            lookup = new BinaryProtoLookupService(this, conf.getServiceUrl(), conf.getListenerName(), conf.isUseTls(),
+            return new BinaryProtoLookupService(this, url, conf.getListenerName(), conf.isUseTls(),
                     externalExecutorProvider.getExecutor());
         }
     }