diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java index 7f7268506e02b..868fcd101f028 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java +++ b/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 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 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 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 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 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 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 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 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 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 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 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 consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java index 28429340bfb9b..27493c9405f6d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java +++ b/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(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java index 580ed3c8c3312..6669d74f2f40b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java +++ b/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())); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java index ea0c87bad50b1..f1483f49c2a87 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java +++ b/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(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java index ba4f7f21da90b..c5c97c990757f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java +++ b/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 authParams = new HashMap<>(); authParams.put("tlsCertFile", TLS_CLIENT_CERT_FILE_PATH); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java index 075c246af40a5..f2a980cae4763 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java +++ b/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 producer = (ProducerImpl) pulsarClient.newProducer().topic(topicName).create(); ClientCnx cnx = producer.cnx(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarMultiHostClientTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarMultiHostClientTest.java new file mode 100644 index 0000000000000..83949902ab5c7 --- /dev/null +++ b/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 consumer = client.newConsumer().topic(topicName).subscriptionName(subscriptionName) + .acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe(); + Producer 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 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 consumer = client.newConsumer().topic(topicName).subscriptionName(subscriptionName) + .acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe(); + Producer 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 msg = null; + Set 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); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java index a44683036433f..718462f387ea3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java +++ b/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(), diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java index f826453799f8c..fca14857f1b32 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java @@ -31,8 +31,6 @@ 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.apache.pulsar.common.policies.data.PersistentTopicInternalStats; import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats.CursorStats; import org.slf4j.Logger; @@ -49,7 +47,6 @@ import static org.mockito.Mockito.spy; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertTrue; -import static org.testng.Assert.fail; public class PulsarBrokerStatsClientTest extends ProducerConsumerBase { @@ -135,63 +132,5 @@ public void testTopicInternalStats() throws Exception { log.info("-- Exiting {} test --", methodName); } - @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 consumer = client.newConsumer().topic(topicName).subscriptionName(subscriptionName) - .acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe(); - Producer 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 producer = client.newProducer().topic(topicName).create(); - - fail(); - } catch (PulsarClientException pce) { - log.error("create producer error: ", pce); - } - - log.info("-- Exiting {} test --", methodName); - } - private static final Logger log = LoggerFactory.getLogger(PulsarBrokerStatsClientTest.class); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java index f053da8608e51..5a2a6dae09f60 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java @@ -41,6 +41,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; @@ -657,9 +658,9 @@ public CompletableFuture getPartitionedTopicMetadata(S TopicName topicName = TopicName.get(topic); AtomicLong opTimeoutMs = new AtomicLong(conf.getOperationTimeoutMs()); Backoff backoff = new BackoffBuilder() - .setInitialTime(100, TimeUnit.NANOSECONDS) + .setInitialTime(100, TimeUnit.MILLISECONDS) .setMandatoryStop(opTimeoutMs.get() * 2, TimeUnit.MILLISECONDS) - .setMax(0, TimeUnit.NANOSECONDS) + .setMax(0, TimeUnit.MILLISECONDS) .create(); getPartitionedTopicMetadata(topicName, backoff, opTimeoutMs, metadataFuture); } catch (IllegalArgumentException e) { @@ -674,12 +675,16 @@ private void getPartitionedTopicMetadata(TopicName topicName, CompletableFuture future) { lookup.getPartitionedTopicMetadata(topicName).thenAccept(future::complete).exceptionally(e -> { long nextDelay = Math.min(backoff.next(), remainingTime.get()); - if (nextDelay <= 0) { + // skip retry scheduler when set lookup throttle in client or server side which will lead to `TooManyRequestsException` + boolean isLookupThrottling = e.getCause() instanceof PulsarClientException.TooManyRequestsException; + if (nextDelay <= 0 || isLookupThrottling) { future.completeExceptionally(e); return null; } - timer.newTimeout( task -> { + ((ScheduledExecutorService) externalExecutorProvider.getExecutor()).schedule(() -> { + log.warn("[topic: {}] Could not get connection while getPartitionedTopicMetadata -- Will try again in {} ms", + topicName, nextDelay); remainingTime.addAndGet(-nextDelay); getPartitionedTopicMetadata(topicName, backoff, remainingTime, future); }, nextDelay, TimeUnit.MILLISECONDS); diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConnectionTimeoutTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConnectionTimeoutTest.java index fb072e573891c..15f46ac3e4a3c 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConnectionTimeoutTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConnectionTimeoutTest.java @@ -18,15 +18,11 @@ */ package org.apache.pulsar.client.impl; -import io.netty.channel.ConnectTimeoutException; - import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; -import org.apache.http.conn.ConnectionPoolTimeoutException; +import io.netty.channel.ConnectTimeoutException; import org.apache.pulsar.client.api.PulsarClient; - -import org.apache.pulsar.client.api.PulsarClientException; import org.testng.Assert; import org.testng.annotations.Test; @@ -40,8 +36,10 @@ public void testLowTimeout() throws Exception { long startNanos = System.nanoTime(); try (PulsarClient clientLow = PulsarClient.builder().serviceUrl(blackholeBroker) - .connectionTimeout(1, TimeUnit.MILLISECONDS).build(); - PulsarClient clientDefault = PulsarClient.builder().serviceUrl(blackholeBroker).build()) { + .connectionTimeout(1, TimeUnit.MILLISECONDS) + .operationTimeout(1000, TimeUnit.MILLISECONDS).build(); + PulsarClient clientDefault = PulsarClient.builder().serviceUrl(blackholeBroker) + .operationTimeout(1000, TimeUnit.MILLISECONDS).build()) { CompletableFuture lowFuture = clientLow.newProducer().topic("foo").createAsync(); CompletableFuture defaultFuture = clientDefault.newProducer().topic("foo").createAsync(); diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyConnectionThrottlingTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyConnectionThrottlingTest.java index 1946ad4be2de7..0cd76c828da8f 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyConnectionThrottlingTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyConnectionThrottlingTest.java @@ -21,6 +21,7 @@ import static org.mockito.Mockito.doReturn; import java.util.Optional; +import java.util.concurrent.TimeUnit; import org.apache.bookkeeper.test.PortManager; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; @@ -73,12 +74,12 @@ protected void cleanup() throws Exception { public void testInboundConnection() throws Exception { LOG.info("Creating producer 1"); PulsarClient client1 = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort().get()) - .build(); + .operationTimeout(1000, TimeUnit.MILLISECONDS).build(); Producer producer1 = client1.newProducer(Schema.BYTES).topic("persistent://sample/test/local/producer-topic-1").create(); LOG.info("Creating producer 2"); PulsarClient client2 = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort().get()) - .build(); + .operationTimeout(1000, TimeUnit.MILLISECONDS).build(); Producer producer2; Assert.assertEquals(ProxyService.rejectedConnections.get(), 0.0d); try { @@ -88,7 +89,8 @@ public void testInboundConnection() throws Exception { } catch (Exception ex) { // OK } - Assert.assertEquals(ProxyService.rejectedConnections.get(), 1.0d); + // should add retry count since retry every 100ms and operation timeout is set to 1000ms + Assert.assertEquals(ProxyService.rejectedConnections.get(), 11.0d); } private static final Logger LOG = LoggerFactory.getLogger(ProxyConnectionThrottlingTest.class); diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyLookupThrottlingTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyLookupThrottlingTest.java index 2c1c691299b2c..5d184a700700d 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyLookupThrottlingTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyLookupThrottlingTest.java @@ -22,6 +22,7 @@ import static org.testng.Assert.assertTrue; import java.util.Optional; +import java.util.concurrent.TimeUnit; import org.apache.bookkeeper.test.PortManager; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; @@ -32,9 +33,7 @@ import org.apache.pulsar.common.configuration.PulsarConfigurationLoader; import org.mockito.Mockito; import org.testng.Assert; -import org.testng.annotations.AfterClass; import org.testng.annotations.AfterMethod; -import org.testng.annotations.BeforeClass; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @@ -75,7 +74,7 @@ protected void cleanup() throws Exception { @Test public void testLookup() throws Exception { PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:" + proxyConfig.getServicePort().get()) - .connectionsPerBroker(5).ioThreads(5).build(); + .connectionsPerBroker(5).ioThreads(5).operationTimeout(1000, TimeUnit.MILLISECONDS).build(); assertTrue(proxyService.getLookupRequestSemaphore().tryAcquire()); assertTrue(proxyService.getLookupRequestSemaphore().tryAcquire()); Producer producer1 = client.newProducer(Schema.BYTES).topic("persistent://sample/test/local/producer-topic") @@ -88,7 +87,7 @@ public void testLookup() throws Exception { } catch (Exception ex) { // Ignore } - Assert.assertEquals(LookupProxyHandler.rejectedPartitionsMetadataRequests.get(), 1.0d); + Assert.assertEquals(LookupProxyHandler.rejectedPartitionsMetadataRequests.get(), 11.0d); proxyService.getLookupRequestSemaphore().release(); try { Producer producer3 = client.newProducer(Schema.BYTES).topic("persistent://sample/test/local/producer-topic") @@ -96,7 +95,7 @@ public void testLookup() throws Exception { } catch (Exception ex) { Assert.fail("Should not have failed since can acquire LookupRequestSemaphore"); } - Assert.assertEquals(LookupProxyHandler.rejectedPartitionsMetadataRequests.get(), 1.0d); + Assert.assertEquals(LookupProxyHandler.rejectedPartitionsMetadataRequests.get(), 11.0d); client.close(); } } diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java index fe206be5c692e..04c7e7a8d082a 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java @@ -474,6 +474,7 @@ private PulsarClient createPulsarClient(String proxyServiceUrl, ClientBuilder cl return clientBuilder.serviceUrl(proxyServiceUrl).statsInterval(0, TimeUnit.SECONDS) .tlsTrustCertsFilePath(TLS_PROXY_TRUST_CERT_FILE_PATH).allowTlsInsecureConnection(true) - .authentication(authTls).enableTls(true).build(); + .authentication(authTls).enableTls(true) + .operationTimeout(1000, TimeUnit.MILLISECONDS).build(); } }