From 4527adc1800ec546e1b406d2ec2711dabd00ed21 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 4 Jan 2024 19:44:23 +0800 Subject: [PATCH 01/12] [fix] [client] Messages lost due to TopicListWatcher reconnect (cherry picked from commit f455b8d3db451759ae6a404a45786c453cf95a48) --- .../auth/MockedPulsarServiceBaseTest.java | 8 ++ .../impl/PatternTopicsConsumerImplTest.java | 63 ++++++++++++--- .../impl/PatternMultiTopicsConsumerImpl.java | 79 ++++++++++++++++--- .../pulsar/client/impl/TopicListWatcher.java | 7 +- .../client/impl/TopicListWatcherTest.java | 2 +- 5 files changed, 133 insertions(+), 26 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java index b8d75bd0fbcac..cc5ea3bbb7bd4 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java @@ -708,5 +708,13 @@ public static class ServiceProducer { private PersistentTopic persistentTopic; } + protected void sleepSeconds(int seconds){ + try { + Thread.currentThread().sleep(1000 * seconds); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + private static final Logger log = LoggerFactory.getLogger(MockedPulsarServiceBaseTest.class); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java index 451f93067b2ca..9115cefa6e1e6 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java @@ -37,14 +37,18 @@ import io.netty.util.Timeout; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.InjectedClientCnxClientBuilder; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageRoutingMode; 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.client.api.RegexSubscriptionMode; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.common.api.proto.BaseCommand; +import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.policies.data.TenantInfoImpl; import org.awaitility.Awaitility; @@ -53,6 +57,7 @@ import org.slf4j.LoggerFactory; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; +import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @Test(groups = "broker-impl") @@ -620,13 +625,28 @@ public void testStartEmptyPatternConsumer() throws Exception { producer3.close(); } - @Test(timeOut = testTimeout) - public void testAutoSubscribePatterConsumerFromBrokerWatcher() throws Exception { - String key = "AutoSubscribePatternConsumer"; - String subscriptionName = "my-ex-subscription-" + key; + @DataProvider(name= "delayTypesOfWatchingTopics") + public Object[][] delayTypesOfWatchingTopics(){ + return new Object[][]{ + {true}, + {false} + }; + } - Pattern pattern = Pattern.compile("persistent://my-property/my-ns/pattern-topic.*"); - Consumer consumer = pulsarClient.newConsumer() + @Test(timeOut = testTimeout, dataProvider = "delayTypesOfWatchingTopics") + public void testAutoSubscribePatterConsumerFromBrokerWatcher(boolean delayWatchingTopics) throws Exception { + final String key = "AutoSubscribePatternConsumer"; + final String subscriptionName = "my-ex-subscription-" + key; + final Pattern pattern = Pattern.compile("persistent://my-property/my-ns/pattern-topic.*"); + + PulsarClient client = null; + if (delayWatchingTopics) { + client = createDelayWatchTopicsClient(); + } else { + client = pulsarClient; + } + + Consumer consumer = client.newConsumer() .topicsPattern(pattern) // Disable automatic discovery. .patternAutoDiscoveryPeriod(1000) @@ -636,12 +656,6 @@ public void testAutoSubscribePatterConsumerFromBrokerWatcher() throws Exception .receiverQueueSize(4) .subscribe(); - // Wait topic list watcher creation. - Awaitility.await().untilAsserted(() -> { - CompletableFuture completableFuture = WhiteboxImpl.getInternalState(consumer, "watcherFuture"); - assertTrue(completableFuture.isDone() && !completableFuture.isCompletedExceptionally()); - }); - // 1. create partition String topicName = "persistent://my-property/my-ns/pattern-topic-1-" + key; TenantInfoImpl tenantInfo = createDefaultTenantInfo(); @@ -657,7 +671,32 @@ public void testAutoSubscribePatterConsumerFromBrokerWatcher() throws Exception assertEquals(((PatternMultiTopicsConsumerImpl) consumer).getPartitionedTopics().size(), 1); }); + // cleanup. consumer.close(); + admin.topics().deletePartitionedTopic(topicName); + } + + private PulsarClient createDelayWatchTopicsClient() throws Exception { + ClientBuilderImpl clientBuilder = (ClientBuilderImpl) PulsarClient.builder().serviceUrl(lookupUrl.toString()); + return InjectedClientCnxClientBuilder.create(clientBuilder, + (conf, eventLoopGroup) -> new ClientCnx(conf, eventLoopGroup) { + public CompletableFuture newWatchTopicList( + BaseCommand command, long requestId) { + // Inject 2 seconds delay when sending command New Watch Topics. + CompletableFuture res = new CompletableFuture<>(); + new Thread(() -> { + sleepSeconds(2); + super.newWatchTopicList(command, requestId).whenComplete((v, ex) -> { + if (ex != null) { + res.completeExceptionally(ex); + } else { + res.complete(v); + } + }); + }).start(); + return res; + } + }); } // simulate subscribe a pattern which has 3 topics, but then matched topic added in. diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java index c6ea6216cc1f4..ca6895c333b8e 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java @@ -31,6 +31,7 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import java.util.regex.Pattern; import java.util.stream.Collectors; import org.apache.pulsar.client.api.Consumer; @@ -50,9 +51,12 @@ public class PatternMultiTopicsConsumerImpl extends MultiTopicsConsumerImpl watcherFuture; + private final CompletableFuture watcherFuture = new CompletableFuture<>(); protected NamespaceName namespaceName; private volatile Timeout recheckPatternTimeout = null; + private volatile AtomicReference retryRecheckPatternTask = new AtomicReference<>(); + private final Backoff retryRecheckPatternTaskBackoff = new BackoffBuilder().setInitialTime(5, TimeUnit.SECONDS) + .setMax(1, TimeUnit.MINUTES).setMandatoryStop(0, TimeUnit.SECONDS).create(); private volatile String topicsHash; public PatternMultiTopicsConsumerImpl(Pattern topicsPattern, @@ -78,11 +82,10 @@ public PatternMultiTopicsConsumerImpl(Pattern topicsPattern, this.topicsChangeListener = new PatternTopicsChangedListener(); this.recheckPatternTimeout = client.timer() .newTimeout(this, Math.max(1, conf.getPatternAutoDiscoveryPeriod()), TimeUnit.SECONDS); - this.watcherFuture = new CompletableFuture<>(); if (subscriptionMode == Mode.PERSISTENT) { long watcherId = client.newTopicListWatcherId(); new TopicListWatcher(topicsChangeListener, client, topicsPattern, watcherId, - namespaceName, topicsHash, watcherFuture); + namespaceName, topicsHash, watcherFuture, () -> recheckTopicsChangeRetryIfFailed()); watcherFuture .thenAccept(__ -> recheckPatternTimeout.cancel()) .exceptionally(ex -> { @@ -105,7 +108,62 @@ public void run(Timeout timeout) throws Exception { if (timeout.isCancelled()) { return; } - client.getLookup().getTopicsUnderNamespace(namespaceName, subscriptionMode, topicsPattern.pattern(), topicsHash) + asyncRecheckTopicsChange().exceptionally(ex -> { + log.warn("[{}] Failed to recheck topics change: {}", topic, ex.getMessage()); + return null; + }).thenAccept(__ -> { + // schedule the next re-check task + this.recheckPatternTimeout = client.timer() + .newTimeout(PatternMultiTopicsConsumerImpl.this, + Math.max(1, conf.getPatternAutoDiscoveryPeriod()), TimeUnit.SECONDS); + }); + } + + private void recheckTopicsChangeRetryIfFailed() { + recheckTopicsChangeRetryIfFailed(null); + } + + private void recheckTopicsChangeRetryIfFailed(Timeout retryTask) { + // This method will be called by A New Call or Timeout scheduled call. + final boolean isNew = (retryTask == null); + // Skip if closed or the task has been cancelled. + if (getState() == State.Closing || getState() == State.Closed + || (retryTask != null && retryTask.isCancelled())) { + retryRecheckPatternTask.compareAndSet(retryTask, null); + return; + } + // Skip the new check if contains a retry task. + Timeout pendingRetryTask = retryRecheckPatternTask.get(); + if (isNew && pendingRetryTask != null) { + return; + } + // Do check. + asyncRecheckTopicsChange().whenComplete((ignore, ex) -> { + if (ex != null) { + log.warn("[{}] Failed to recheck topics change: {}", topic, ex.getMessage()); + long delayMs = retryRecheckPatternTaskBackoff.next(); + Timeout newTask = client.timer().newTimeout(timeout -> { + if (timeout.cancel()) { + return; + } + recheckTopicsChangeRetryIfFailed(); + }, delayMs, TimeUnit.MILLISECONDS); + if (!retryRecheckPatternTask.compareAndSet(retryTask, newTask)) { + // Another thread added a new task, so cancel current one. + newTask.cancel(); + } + } else { + retryRecheckPatternTaskBackoff.reset(); + if (!isNew) { + retryRecheckPatternTask.compareAndSet(retryTask, null); + } + } + }); + } + + private CompletableFuture asyncRecheckTopicsChange() { + String pattern = topicsPattern.pattern(); + return client.getLookup().getTopicsUnderNamespace(namespaceName, subscriptionMode, pattern, topicsHash) .thenCompose(getTopicsResult -> { if (log.isDebugEnabled()) { @@ -125,14 +183,6 @@ public void run(Timeout timeout) throws Exception { } return updateSubscriptions(topicsPattern, this::setTopicsHash, getTopicsResult, topicsChangeListener, oldTopics); - }).exceptionally(ex -> { - log.warn("[{}] Failed to recheck topics change: {}", topic, ex.getMessage()); - return null; - }).thenAccept(__ -> { - // schedule the next re-check task - this.recheckPatternTimeout = client.timer() - .newTimeout(PatternMultiTopicsConsumerImpl.this, - Math.max(1, conf.getPatternAutoDiscoveryPeriod()), TimeUnit.SECONDS); }); } @@ -234,6 +284,11 @@ public CompletableFuture closeAsync() { timeout.cancel(); recheckPatternTimeout = null; } + Timeout retryTaskToRecheckTopics = retryRecheckPatternTask.get(); + if (retryTaskToRecheckTopics != null) { + retryTaskToRecheckTopics.cancel(); + retryRecheckPatternTask.compareAndSet(retryTaskToRecheckTopics, null); + } List> closeFutures = new ArrayList<>(2); if (watcherFuture.isDone() && !watcherFuture.isCompletedExceptionally()) { TopicListWatcher watcher = watcherFuture.getNow(null); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java index 2ce784dbaac04..489a07a606eb2 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java @@ -56,11 +56,14 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler. private final List previousExceptions = new CopyOnWriteArrayList<>(); private final AtomicReference clientCnxUsedForWatcherRegistration = new AtomicReference<>(); + private final Runnable recheckTopicsChangeAfterReconnect; + public TopicListWatcher(PatternMultiTopicsConsumerImpl.TopicsChangedListener topicsChangeListener, PulsarClientImpl client, Pattern topicsPattern, long watcherId, NamespaceName namespace, String topicsHash, - CompletableFuture watcherFuture) { + CompletableFuture watcherFuture, + Runnable recheckTopicsChangeAfterReconnect) { super(client, topicsPattern.pattern()); this.topicsChangeListener = topicsChangeListener; this.name = "Watcher(" + topicsPattern + ")"; @@ -77,6 +80,7 @@ public TopicListWatcher(PatternMultiTopicsConsumerImpl.TopicsChangedListener top this.namespace = namespace; this.topicsHash = topicsHash; this.watcherFuture = watcherFuture; + this.recheckTopicsChangeAfterReconnect = recheckTopicsChangeAfterReconnect; connectionHandler.grabCnx(); } @@ -141,6 +145,7 @@ public CompletableFuture connectionOpened(ClientCnx cnx) { this.connectionHandler.resetBackoff(); + recheckTopicsChangeAfterReconnect.run(); watcherFuture.complete(this); future.complete(null); }).exceptionally((e) -> { diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TopicListWatcherTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TopicListWatcherTest.java index dd75770b5688d..7e9fd601d4f67 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TopicListWatcherTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TopicListWatcherTest.java @@ -71,7 +71,7 @@ public void setup() { watcherFuture = new CompletableFuture<>(); watcher = new TopicListWatcher(listener, client, Pattern.compile(topic), 7, - NamespaceName.get("tenant/ns"), null, watcherFuture); + NamespaceName.get("tenant/ns"), null, watcherFuture, () -> {}); } @Test From ed1d224e9a0996533de7abfaee5d9358eb1a16f8 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Fri, 5 Jan 2024 18:15:52 +0800 Subject: [PATCH 02/12] - --- .../pulsar/client/impl/PatternMultiTopicsConsumerImpl.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java index ca6895c333b8e..842b643be040a 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java @@ -124,15 +124,16 @@ private void recheckTopicsChangeRetryIfFailed() { } private void recheckTopicsChangeRetryIfFailed(Timeout retryTask) { - // This method will be called by A New Call or Timeout scheduled call. - final boolean isNew = (retryTask == null); // Skip if closed or the task has been cancelled. if (getState() == State.Closing || getState() == State.Closed || (retryTask != null && retryTask.isCancelled())) { retryRecheckPatternTask.compareAndSet(retryTask, null); return; } + // If the argument "retryTask" is not null, it means this method was called by the timer. Otherwise, it is a + // new call. // Skip the new check if contains a retry task. + final boolean isNew = (retryTask == null); Timeout pendingRetryTask = retryRecheckPatternTask.get(); if (isNew && pendingRetryTask != null) { return; From 495fb3834a2c0109ae63e4ef8be32d4d43a584ef Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Fri, 5 Jan 2024 18:19:39 +0800 Subject: [PATCH 03/12] address comments --- .../pulsar/client/impl/PatternMultiTopicsConsumerImpl.java | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java index 842b643be040a..946846096fab3 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java @@ -144,10 +144,7 @@ private void recheckTopicsChangeRetryIfFailed(Timeout retryTask) { log.warn("[{}] Failed to recheck topics change: {}", topic, ex.getMessage()); long delayMs = retryRecheckPatternTaskBackoff.next(); Timeout newTask = client.timer().newTimeout(timeout -> { - if (timeout.cancel()) { - return; - } - recheckTopicsChangeRetryIfFailed(); + recheckTopicsChangeRetryIfFailed(timeout); }, delayMs, TimeUnit.MILLISECONDS); if (!retryRecheckPatternTask.compareAndSet(retryTask, newTask)) { // Another thread added a new task, so cancel current one. From ed1eaec0944cff6a38989a57dc634c91024dd826 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 8 Jan 2024 16:23:26 +0800 Subject: [PATCH 04/12] simplify logic --- .../impl/PatternMultiTopicsConsumerImpl.java | 68 ++++++------------- 1 file changed, 20 insertions(+), 48 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java index 946846096fab3..3b3e8c9265dcc 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java @@ -31,7 +31,6 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; import java.util.regex.Pattern; import java.util.stream.Collectors; import org.apache.pulsar.client.api.Consumer; @@ -54,7 +53,6 @@ public class PatternMultiTopicsConsumerImpl extends MultiTopicsConsumerImpl watcherFuture = new CompletableFuture<>(); protected NamespaceName namespaceName; private volatile Timeout recheckPatternTimeout = null; - private volatile AtomicReference retryRecheckPatternTask = new AtomicReference<>(); private final Backoff retryRecheckPatternTaskBackoff = new BackoffBuilder().setInitialTime(5, TimeUnit.SECONDS) .setMax(1, TimeUnit.MINUTES).setMandatoryStop(0, TimeUnit.SECONDS).create(); private volatile String topicsHash; @@ -85,7 +83,7 @@ public PatternMultiTopicsConsumerImpl(Pattern topicsPattern, if (subscriptionMode == Mode.PERSISTENT) { long watcherId = client.newTopicListWatcherId(); new TopicListWatcher(topicsChangeListener, client, topicsPattern, watcherId, - namespaceName, topicsHash, watcherFuture, () -> recheckTopicsChangeRetryIfFailed()); + namespaceName, topicsHash, watcherFuture, () -> recheckTopicsChangeAfterReconnect()); watcherFuture .thenAccept(__ -> recheckPatternTimeout.cancel()) .exceptionally(ex -> { @@ -102,6 +100,25 @@ public static NamespaceName getNameSpaceFromPattern(Pattern pattern) { return TopicName.get(pattern.pattern()).getNamespaceObject(); } + private void recheckTopicsChangeAfterReconnect() { + // Skip if closed or the task has been cancelled. + if (getState() == State.Closing || getState() == State.Closed) { + return; + } + // Do check. + asyncRecheckTopicsChange().whenComplete((ignore, ex) -> { + if (ex != null) { + log.warn("[{}] Failed to recheck topics change: {}", topic, ex.getMessage()); + long delayMs = retryRecheckPatternTaskBackoff.next(); + client.timer().newTimeout(timeout -> { + recheckTopicsChangeAfterReconnect(); + }, delayMs, TimeUnit.MILLISECONDS); + } else { + retryRecheckPatternTaskBackoff.reset(); + } + }); + } + // TimerTask to recheck topics change, and trigger subscribe/unsubscribe based on the change. @Override public void run(Timeout timeout) throws Exception { @@ -119,46 +136,6 @@ public void run(Timeout timeout) throws Exception { }); } - private void recheckTopicsChangeRetryIfFailed() { - recheckTopicsChangeRetryIfFailed(null); - } - - private void recheckTopicsChangeRetryIfFailed(Timeout retryTask) { - // Skip if closed or the task has been cancelled. - if (getState() == State.Closing || getState() == State.Closed - || (retryTask != null && retryTask.isCancelled())) { - retryRecheckPatternTask.compareAndSet(retryTask, null); - return; - } - // If the argument "retryTask" is not null, it means this method was called by the timer. Otherwise, it is a - // new call. - // Skip the new check if contains a retry task. - final boolean isNew = (retryTask == null); - Timeout pendingRetryTask = retryRecheckPatternTask.get(); - if (isNew && pendingRetryTask != null) { - return; - } - // Do check. - asyncRecheckTopicsChange().whenComplete((ignore, ex) -> { - if (ex != null) { - log.warn("[{}] Failed to recheck topics change: {}", topic, ex.getMessage()); - long delayMs = retryRecheckPatternTaskBackoff.next(); - Timeout newTask = client.timer().newTimeout(timeout -> { - recheckTopicsChangeRetryIfFailed(timeout); - }, delayMs, TimeUnit.MILLISECONDS); - if (!retryRecheckPatternTask.compareAndSet(retryTask, newTask)) { - // Another thread added a new task, so cancel current one. - newTask.cancel(); - } - } else { - retryRecheckPatternTaskBackoff.reset(); - if (!isNew) { - retryRecheckPatternTask.compareAndSet(retryTask, null); - } - } - }); - } - private CompletableFuture asyncRecheckTopicsChange() { String pattern = topicsPattern.pattern(); return client.getLookup().getTopicsUnderNamespace(namespaceName, subscriptionMode, pattern, topicsHash) @@ -282,11 +259,6 @@ public CompletableFuture closeAsync() { timeout.cancel(); recheckPatternTimeout = null; } - Timeout retryTaskToRecheckTopics = retryRecheckPatternTask.get(); - if (retryTaskToRecheckTopics != null) { - retryTaskToRecheckTopics.cancel(); - retryRecheckPatternTask.compareAndSet(retryTaskToRecheckTopics, null); - } List> closeFutures = new ArrayList<>(2); if (watcherFuture.isDone() && !watcherFuture.isCompletedExceptionally()) { TopicListWatcher watcher = watcherFuture.getNow(null); From 34eeff20292831b15cb3747eadca1b4f572218a4 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 8 Jan 2024 16:51:41 +0800 Subject: [PATCH 05/12] address comments --- .../impl/PatternMultiTopicsConsumerImpl.java | 30 ++++++++++++++----- 1 file changed, 23 insertions(+), 7 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java index 3b3e8c9265dcc..a9b455240a6bd 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java @@ -52,9 +52,17 @@ public class PatternMultiTopicsConsumerImpl extends MultiTopicsConsumerImpl watcherFuture = new CompletableFuture<>(); protected NamespaceName namespaceName; + + /** + * There is two task to re-check topic changes, the both tasks will not be take affects at the same time. + * 1. {@link #recheckTopicsChangeAfterReconnect}: it will be called after the {@link TopicListWatcher} reconnected + * if you enabled {@link TopicListWatcher}. This backoff used to do a retry if + * {@link #recheckTopicsChangeAfterReconnect} is failed. + * 2. {@link #run} A scheduled task to trigger re-check topic changes, it will be used if you disabled + * {@link TopicListWatcher}. + */ + private final Backoff recheckPatternTaskBackoffIfFailed; private volatile Timeout recheckPatternTimeout = null; - private final Backoff retryRecheckPatternTaskBackoff = new BackoffBuilder().setInitialTime(5, TimeUnit.SECONDS) - .setMax(1, TimeUnit.MINUTES).setMandatoryStop(0, TimeUnit.SECONDS).create(); private volatile String topicsHash; public PatternMultiTopicsConsumerImpl(Pattern topicsPattern, @@ -71,6 +79,11 @@ public PatternMultiTopicsConsumerImpl(Pattern topicsPattern, this.topicsPattern = topicsPattern; this.topicsHash = topicsHash; this.subscriptionMode = subscriptionMode; + this.recheckPatternTaskBackoffIfFailed = new BackoffBuilder() + .setInitialTime(client.getConfiguration().getInitialBackoffIntervalNanos(), TimeUnit.NANOSECONDS) + .setMax(client.getConfiguration().getMaxBackoffIntervalNanos(), TimeUnit.NANOSECONDS) + .setMandatoryStop(0, TimeUnit.SECONDS) + .create(); if (this.namespaceName == null) { this.namespaceName = getNameSpaceFromPattern(topicsPattern); @@ -100,21 +113,24 @@ public static NamespaceName getNameSpaceFromPattern(Pattern pattern) { return TopicName.get(pattern.pattern()).getNamespaceObject(); } + /** + * This method will be called after the {@link TopicListWatcher} reconnected after enabled {@link TopicListWatcher}. + */ private void recheckTopicsChangeAfterReconnect() { // Skip if closed or the task has been cancelled. if (getState() == State.Closing || getState() == State.Closed) { return; } // Do check. - asyncRecheckTopicsChange().whenComplete((ignore, ex) -> { + recheckTopicsChange().whenComplete((ignore, ex) -> { if (ex != null) { log.warn("[{}] Failed to recheck topics change: {}", topic, ex.getMessage()); - long delayMs = retryRecheckPatternTaskBackoff.next(); + long delayMs = recheckPatternTaskBackoffIfFailed.next(); client.timer().newTimeout(timeout -> { recheckTopicsChangeAfterReconnect(); }, delayMs, TimeUnit.MILLISECONDS); } else { - retryRecheckPatternTaskBackoff.reset(); + recheckPatternTaskBackoffIfFailed.reset(); } }); } @@ -125,7 +141,7 @@ public void run(Timeout timeout) throws Exception { if (timeout.isCancelled()) { return; } - asyncRecheckTopicsChange().exceptionally(ex -> { + recheckTopicsChange().exceptionally(ex -> { log.warn("[{}] Failed to recheck topics change: {}", topic, ex.getMessage()); return null; }).thenAccept(__ -> { @@ -136,7 +152,7 @@ public void run(Timeout timeout) throws Exception { }); } - private CompletableFuture asyncRecheckTopicsChange() { + private CompletableFuture recheckTopicsChange() { String pattern = topicsPattern.pattern(); return client.getLookup().getTopicsUnderNamespace(namespaceName, subscriptionMode, pattern, topicsHash) .thenCompose(getTopicsResult -> { From 78971f25f2f7196f297bc850f388ac47d667ec34 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 8 Jan 2024 16:55:32 +0800 Subject: [PATCH 06/12] address comments --- .../apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java | 2 +- .../pulsar/client/impl/PatternTopicsConsumerImplTest.java | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java index cc5ea3bbb7bd4..30e48caf97058 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java @@ -712,7 +712,7 @@ protected void sleepSeconds(int seconds){ try { Thread.currentThread().sleep(1000 * seconds); } catch (InterruptedException e) { - throw new RuntimeException(e); + Thread.currentThread().interrupt(); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java index 9115cefa6e1e6..c708b4cae0a19 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java @@ -674,6 +674,9 @@ public void testAutoSubscribePatterConsumerFromBrokerWatcher(boolean delayWatchi // cleanup. consumer.close(); admin.topics().deletePartitionedTopic(topicName); + if (delayWatchingTopics) { + client.close(); + } } private PulsarClient createDelayWatchTopicsClient() throws Exception { From 42cda95eaa1210dd3e461d263d037d5aae6d8908 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 8 Jan 2024 16:56:23 +0800 Subject: [PATCH 07/12] address comments --- .../apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java index 30e48caf97058..c3909161a9140 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java @@ -712,6 +712,7 @@ protected void sleepSeconds(int seconds){ try { Thread.currentThread().sleep(1000 * seconds); } catch (InterruptedException e) { + log.warn("This thread has been interrupted", e); Thread.currentThread().interrupt(); } } From 7616f7ccc5082a86ce4aaef523711245291c9533 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 8 Jan 2024 16:58:09 +0800 Subject: [PATCH 08/12] address comments --- .../client/impl/PatternMultiTopicsConsumerImpl.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java index a9b455240a6bd..208e1fa05c7b8 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java @@ -61,7 +61,7 @@ public class PatternMultiTopicsConsumerImpl extends MultiTopicsConsumerImpl { if (ex != null) { log.warn("[{}] Failed to recheck topics change: {}", topic, ex.getMessage()); - long delayMs = recheckPatternTaskBackoffIfFailed.next(); + long delayMs = recheckPatternTaskBackoff.next(); client.timer().newTimeout(timeout -> { recheckTopicsChangeAfterReconnect(); }, delayMs, TimeUnit.MILLISECONDS); } else { - recheckPatternTaskBackoffIfFailed.reset(); + recheckPatternTaskBackoff.reset(); } }); } From fb40dd76eb4b564b8f716449a1fa30dcec0035d8 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 8 Jan 2024 17:24:16 +0800 Subject: [PATCH 09/12] address comments --- .../apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java index c3909161a9140..eb75963061edd 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java @@ -710,7 +710,7 @@ public static class ServiceProducer { protected void sleepSeconds(int seconds){ try { - Thread.currentThread().sleep(1000 * seconds); + Thread.sleep(1000 * seconds); } catch (InterruptedException e) { log.warn("This thread has been interrupted", e); Thread.currentThread().interrupt(); From 59e7ef98921e14a9e836359bb97fdc6f40b38bde Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 8 Jan 2024 17:41:35 +0800 Subject: [PATCH 10/12] fix the issue of thread-safety --- .../impl/PatternMultiTopicsConsumerImpl.java | 38 +++++++++++-------- 1 file changed, 23 insertions(+), 15 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java index 208e1fa05c7b8..c7e908f4a998d 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java @@ -31,6 +31,7 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Pattern; import java.util.stream.Collectors; import org.apache.pulsar.client.api.Consumer; @@ -62,6 +63,7 @@ public class PatternMultiTopicsConsumerImpl extends MultiTopicsConsumerImpl recheckTopicsChange() { String pattern = topicsPattern.pattern(); + final int epoch = recheckPatternEpoch.incrementAndGet(); return client.getLookup().getTopicsUnderNamespace(namespaceName, subscriptionMode, pattern, topicsHash) .thenCompose(getTopicsResult -> { + // If "recheckTopicsChange" has been called more than one times, only make the last one take affects. + synchronized (PatternMultiTopicsConsumerImpl.this) { + if (recheckPatternEpoch.get() > epoch) { + return CompletableFuture.completedFuture(null); + } + if (log.isDebugEnabled()) { + log.debug("Get topics under namespace {}, topics.size: {}, topicsHash: {}, filtered: {}", + namespaceName, getTopicsResult.getTopics().size(), getTopicsResult.getTopicsHash(), + getTopicsResult.isFiltered()); + getTopicsResult.getTopics().forEach(topicName -> + log.debug("Get topics under namespace {}, topic: {}", namespaceName, topicName)); + } - if (log.isDebugEnabled()) { - log.debug("Get topics under namespace {}, topics.size: {}, topicsHash: {}, filtered: {}", - namespaceName, getTopicsResult.getTopics().size(), getTopicsResult.getTopicsHash(), - getTopicsResult.isFiltered()); - getTopicsResult.getTopics().forEach(topicName -> - log.debug("Get topics under namespace {}, topic: {}", namespaceName, topicName)); - } - - final List oldTopics = new ArrayList<>(getPartitionedTopics()); - for (String partition : getPartitions()) { - TopicName topicName = TopicName.get(partition); - if (!topicName.isPartitioned() || !oldTopics.contains(topicName.getPartitionedTopicName())) { - oldTopics.add(partition); + final List oldTopics = new ArrayList<>(getPartitionedTopics()); + for (String partition : getPartitions()) { + TopicName topicName = TopicName.get(partition); + if (!topicName.isPartitioned() || !oldTopics.contains(topicName.getPartitionedTopicName())) { + oldTopics.add(partition); + } } + return updateSubscriptions(topicsPattern, this::setTopicsHash, getTopicsResult, + topicsChangeListener, oldTopics); } - return updateSubscriptions(topicsPattern, this::setTopicsHash, getTopicsResult, - topicsChangeListener, oldTopics); }); } From aec1eba7dfa3a7e58b6a7cb400618de950584203 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 8 Jan 2024 17:44:11 +0800 Subject: [PATCH 11/12] seperate the lockj --- .../pulsar/client/impl/PatternMultiTopicsConsumerImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java index c7e908f4a998d..82f8646787866 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java @@ -160,7 +160,7 @@ private CompletableFuture recheckTopicsChange() { return client.getLookup().getTopicsUnderNamespace(namespaceName, subscriptionMode, pattern, topicsHash) .thenCompose(getTopicsResult -> { // If "recheckTopicsChange" has been called more than one times, only make the last one take affects. - synchronized (PatternMultiTopicsConsumerImpl.this) { + synchronized (recheckPatternEpoch) { if (recheckPatternEpoch.get() > epoch) { return CompletableFuture.completedFuture(null); } From 14286042f2bd589a7e0d76dbb6e57c7bf2b035b1 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 8 Jan 2024 17:55:11 +0800 Subject: [PATCH 12/12] seperate the lock --- .../pulsar/client/impl/PatternMultiTopicsConsumerImpl.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java index 82f8646787866..f3ebcdee6c0d9 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java @@ -160,7 +160,9 @@ private CompletableFuture recheckTopicsChange() { return client.getLookup().getTopicsUnderNamespace(namespaceName, subscriptionMode, pattern, topicsHash) .thenCompose(getTopicsResult -> { // If "recheckTopicsChange" has been called more than one times, only make the last one take affects. - synchronized (recheckPatternEpoch) { + // Use "synchronized (recheckPatternTaskBackoff)" instead of + // `synchronized(PatternMultiTopicsConsumerImpl.this)` to avoid locking in a wider range. + synchronized (recheckPatternTaskBackoff) { if (recheckPatternEpoch.get() > epoch) { return CompletableFuture.completedFuture(null); }