[Reopen][Issue 5597] Retry when getPartitionedTopicMetadata failed

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java

@@ -429,7 +429,7 @@ public void testTlsDisabled() throws Exception {
         // Case 1: Access without TLS
         try {
             pulsarClient = PulsarClient.builder().serviceUrl(brokerUrl.toString()).statsInterval(0, TimeUnit.SECONDS)
-                    .build();
+                    .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
             @Cleanup
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
                     .subscribe();
@@ -442,7 +442,8 @@ public void testTlsDisabled() throws Exception {
         // Case 2: Access with TLS
         try {
             pulsarClient = PulsarClient.builder().serviceUrl(brokerUrlTls.toString()).enableTls(true)
-                    .statsInterval(0, TimeUnit.SECONDS).build();
+                    .statsInterval(0, TimeUnit.SECONDS)
+                    .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
 
             @Cleanup
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
@@ -472,7 +473,7 @@ public void testTlsEnabled() throws Exception {
         PulsarClient pulsarClient = null;
         try {
             pulsarClient = PulsarClient.builder().serviceUrl(brokerUrl.toString()).statsInterval(0, TimeUnit.SECONDS)
-                    .build();
+                    .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
             @Cleanup
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
                     .subscribe();
@@ -485,7 +486,8 @@ public void testTlsEnabled() throws Exception {
         // Case 2: Access with TLS (Allow insecure TLS connection)
         try {
             pulsarClient = PulsarClient.builder().serviceUrl(brokerUrlTls.toString()).enableTls(true)
-                    .allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS).build();
+                    .allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS)
+                    .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
 
             @Cleanup
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
@@ -500,7 +502,8 @@ public void testTlsEnabled() throws Exception {
         // Case 3: Access with TLS (Disallow insecure TLS connection)
         try {
             pulsarClient = PulsarClient.builder().serviceUrl(brokerUrlTls.toString()).enableTls(true)
-                    .allowTlsInsecureConnection(false).statsInterval(0, TimeUnit.SECONDS).build();
+                    .allowTlsInsecureConnection(false).statsInterval(0, TimeUnit.SECONDS)
+                    .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
 
             @Cleanup
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
@@ -517,7 +520,8 @@ public void testTlsEnabled() throws Exception {
         try {
             pulsarClient = PulsarClient.builder().serviceUrl(brokerUrlTls.toString()).enableTls(true)
                     .allowTlsInsecureConnection(false).tlsTrustCertsFilePath(TLS_SERVER_CERT_FILE_PATH)
-                    .statsInterval(0, TimeUnit.SECONDS).build();
+                    .statsInterval(0, TimeUnit.SECONDS)
+                    .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
 
             @Cleanup
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
@@ -557,7 +561,8 @@ public void testTlsAuthAllowInsecure() throws Exception {
         // Case 1: Access without client certificate
         try {
             pulsarClient = PulsarClient.builder().serviceUrl(brokerUrlTls.toString()).enableTls(true)
-                    .allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS).build();
+                    .allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS)
+                    .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
 
             @Cleanup
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
@@ -576,7 +581,8 @@ public void testTlsAuthAllowInsecure() throws Exception {
             auth.configure(authParams);
 
             pulsarClient = PulsarClient.builder().authentication(auth).serviceUrl(brokerUrlTls.toString())
-                    .enableTls(true).allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS).build();
+                    .enableTls(true).allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS)
+                    .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
 
             @Cleanup
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
@@ -617,7 +623,8 @@ public void testTlsAuthDisallowInsecure() throws Exception {
         // Case 1: Access without client certificate
         try {
             pulsarClient = PulsarClient.builder().serviceUrl(brokerUrlTls.toString()).enableTls(true)
-                    .allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS).build();
+                    .allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS)
+                    .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
 
             @Cleanup
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
@@ -635,7 +642,8 @@ public void testTlsAuthDisallowInsecure() throws Exception {
             auth = new AuthenticationTls();
             auth.configure(authParams);
             pulsarClient = PulsarClient.builder().authentication(auth).serviceUrl(brokerUrlTls.toString())
-                    .enableTls(true).allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS).build();
+                    .enableTls(true).allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS)
+                    .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
 
             @Cleanup
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
@@ -677,7 +685,8 @@ public void testTlsAuthUseTrustCert() throws Exception {
         // Case 1: Access without client certificate
         try {
             pulsarClient = PulsarClient.builder().serviceUrl(brokerUrlTls.toString()).enableTls(true)
-                    .allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS).build();
+                    .allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS)
+                    .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
 
             @Cleanup
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
@@ -694,7 +703,8 @@ public void testTlsAuthUseTrustCert() throws Exception {
             auth = new AuthenticationTls();
             auth.configure(authParams);
             pulsarClient = PulsarClient.builder().authentication(auth).serviceUrl(brokerUrlTls.toString())
-                    .enableTls(true).allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS).build();
+                    .enableTls(true).allowTlsInsecureConnection(true).statsInterval(0, TimeUnit.SECONDS)
+                    .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
 
             @Cleanup
             Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java

@@ -107,7 +107,8 @@ public void testPeerClusterTopicLookup(String protocol) throws Exception {
         final String topic1 = "persistent://" + namespace1 + "/topic1";
         final String topic2 = "persistent://" + namespace2 + "/topic2";
 
-        PulsarClient client3 = PulsarClient.builder().serviceUrl(serviceUrl).statsInterval(0, TimeUnit.SECONDS).build();
+        PulsarClient client3 = PulsarClient.builder().serviceUrl(serviceUrl).statsInterval(0, TimeUnit.SECONDS)
+            .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
         try {
             // try to create producer for topic1 (part of cluster: r1) by calling cluster: r3
             client3.newProducer().topic(topic1).create();

pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java

@@ -29,6 +29,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 
 import javax.naming.AuthenticationException;
 
@@ -119,10 +120,12 @@ public void testProducerAndConsumerAuthorization() throws Exception {
         Authentication authenticationInvalidRole = new ClientAuthentication("test-role");
 
         @Cleanup
-        PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(lookupUrl).authentication(authentication).build();
+        PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(lookupUrl).authentication(authentication)
+                .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
 
         @Cleanup
         PulsarClient pulsarClientInvalidRole = PulsarClient.builder().serviceUrl(lookupUrl)
+                .operationTimeout(1000, TimeUnit.MILLISECONDS)
                 .authentication(authenticationInvalidRole).build();
 
         admin.clusters().createCluster("test", new ClusterData(brokerUrl.toString()));

pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java

@@ -722,7 +722,8 @@ public void start() throws PulsarClientException {
         };
 
         @Cleanup
-        PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(discoverySvcUrl).authentication(auth).build();
+        PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(discoverySvcUrl).authentication(auth)
+            .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
         try {
             pulsarClient.newConsumer().topic("persistent://my-property/use2/my-ns/my-topic1")
                     .subscriptionName("my-subscriber-name").subscribe();
@@ -783,7 +784,8 @@ public void start() throws PulsarClientException {
         };
 
         @Cleanup
-        PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(discoverySvcUrl).authentication(auth).build();
+        PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(discoverySvcUrl).authentication(auth)
+            .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
         try {
             pulsarClient.newConsumer().topic("persistent://my-property/use2/my-ns/my-topic1")
                     .subscriptionName("my-subscriber-name").subscribe();

pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java

@@ -24,6 +24,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.impl.auth.AuthenticationTls;
@@ -77,7 +78,8 @@ protected void internalSetUpForClient(boolean addCertificates, String lookupUrl)
         }
 
         ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(lookupUrl)
-                .tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH).enableTls(true).allowTlsInsecureConnection(false);
+                .tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH).enableTls(true).allowTlsInsecureConnection(false)
+                .operationTimeout(1000, TimeUnit.MILLISECONDS);
         if (addCertificates) {
             Map<String, String> authParams = new HashMap<>();
             authParams.put("tlsCertFile", TLS_CLIENT_CERT_FILE_PATH);

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java

@@ -656,7 +656,8 @@ public void testCloseConnectionOnInternalServerError() throws Exception {
         final String topicName = "persistent://prop/usw/my-ns/newTopic";
 
         String lookupUrl = new URI("pulsar://localhost:" + BROKER_PORT).toString();
-        pulsarClient = PulsarClient.builder().serviceUrl(lookupUrl).statsInterval(0, TimeUnit.SECONDS).build();
+        pulsarClient = PulsarClient.builder().serviceUrl(lookupUrl).statsInterval(0, TimeUnit.SECONDS)
+            .operationTimeout(1000, TimeUnit.MILLISECONDS).build();
 
         ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer().topic(topicName).create();
         ClientCnx cnx = producer.cnx();

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarMultiHostClientTest.java

@@ -0,0 +1,157 @@
+/**
+ * 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.client.impl;
+
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import static org.testng.Assert.fail;
+
+public class PulsarMultiHostClientTest extends ProducerConsumerBase {
+
+    private static final Logger log = LoggerFactory.getLogger(PulsarMultiHostClientTest.class);
+
+    @BeforeMethod
+    @Override
+    protected void setup() throws Exception {
+        super.internalSetup();
+        super.producerBaseSetup();
+    }
+
+    @AfterMethod
+    @Override
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    @Test
+    public void testGetPartitionedTopicMetaData() throws Exception {
+        log.info("-- Starting {} test --", methodName);
+
+        final String topicName = "persistent://my-property/my-ns/my-topic1";
+        final String subscriptionName = "my-subscriber-name";
+
+        try {
+            String url = "http://localhost:" + BROKER_WEBSERVICE_PORT;
+            if (isTcpLookup) {
+                url = "pulsar://localhost:" + BROKER_PORT;
+            }
+            PulsarClient client = newPulsarClient(url, 0);
+
+            Consumer<byte[]> consumer = client.newConsumer().topic(topicName).subscriptionName(subscriptionName)
+                .acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe();
+            Producer<byte[]> producer = client.newProducer().topic(topicName).create();
+
+            consumer.close();
+            producer.close();
+            client.close();
+        } catch (PulsarClientException pce) {
+            log.error("create producer or consumer error: ", pce);
+            fail();
+        }
+
+        log.info("-- Exiting {} test --", methodName);
+    }
+
+    @Test (timeOut = 4000)
+    public void testGetPartitionedTopicDataTimeout() {
+        log.info("-- Starting {} test --", methodName);
+
+        final String topicName = "persistent://my-property/my-ns/my-topic1";
+
+        String url = "http://localhost:51000,localhost:51001";
+        if (isTcpLookup) {
+            url = "pulsar://localhost:51000,localhost:51001";
+        }
+
+        PulsarClient client;
+        try {
+            client = PulsarClient.builder()
+                .serviceUrl(url)
+                .statsInterval(0, TimeUnit.SECONDS)
+                .operationTimeout(3, TimeUnit.SECONDS)
+                .build();
+
+            Producer<byte[]> producer = client.newProducer().topic(topicName).create();
+
+            fail();
+        } catch (PulsarClientException pce) {
+            log.error("create producer error: ", pce);
+        }
+
+        log.info("-- Exiting {} test --", methodName);
+    }
+
+    @Test
+    public void testMultiHostUrlRetrySuccess() throws Exception {
+        log.info("-- Starting {} test --", methodName);
+
+        final String topicName = "persistent://my-property/my-ns/my-topic1";
+        final String subscriptionName = "my-subscriber-name";
+
+        // Multi hosts included an unreached port and the actual port for verify retry logic
+        String urlsWithUnreached = "http://localhost:51000,localhost:" + BROKER_WEBSERVICE_PORT;
+        if (isTcpLookup) {
+            urlsWithUnreached = "pulsar://localhost:51000,localhost:" + BROKER_PORT;
+        }
+        PulsarClient client = newPulsarClient(urlsWithUnreached, 0);
+
+        Consumer<byte[]> consumer = client.newConsumer().topic(topicName).subscriptionName(subscriptionName)
+            .acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe();
+        Producer<byte[]> producer = client.newProducer().topic(topicName).create();
+
+        for (int i = 0; i < 5; i++) {
+            String message = "my-message-" + i;
+            producer.send(message.getBytes());
+            log.info("Produced message: [{}]", message);
+        }
+
+        Message<byte[]> msg = null;
+        Set<String> messageSet = new HashSet();
+        for (int i = 0; i < 5; i++) {
+            msg = consumer.receive(5, TimeUnit.SECONDS);
+            String receivedMessage = new String(msg.getData());
+            log.info("Received message: [{}]", receivedMessage);
+            String expectedMessage = "my-message-" + i;
+            testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage);
+        }
+
+        // Acknowledge the consumption of all messages at once
+        consumer.acknowledgeCumulative(msg);
+        consumer.close();
+
+        producer.close();
+        client.close();
+
+        log.info("-- Exiting {} test --", methodName);
+    }
+}

pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java

@@ -179,7 +179,8 @@ void setup(Method method) throws Exception {
         ClusterData clusterData = new ClusterData(brokerWebServiceUrl.toString());
         superUserAdmin.clusters().updateCluster(config.getClusterName(), clusterData);
 
-        ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl());
+        ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl())
+                .operationTimeout(1000, TimeUnit.MILLISECONDS);
         if (isNotBlank(workerConfig.getClientAuthenticationPlugin())
                 && isNotBlank(workerConfig.getClientAuthenticationParameters())) {
             clientBuilder.authentication(workerConfig.getClientAuthenticationPlugin(),