[fix] [client] Messages lost due to TopicListWatcher reconnect

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java

@@ -397,7 +397,8 @@ private void updateTopicPolicyByBrokerConfig() {
         topicPolicies.getSchemaValidationEnforced().updateBrokerValue(config.isSchemaValidationEnforced());
         topicPolicies.getEntryFilters().updateBrokerValue(new EntryFilters(String.join(",",
                 config.getEntryFilterNames())));
-
+        topicPolicies.getDispatcherPauseOnAckStatePersistentEnabled()
+                .updateBrokerValue(config.isDispatcherPauseOnAckStatePersistentEnabled());
         updateEntryFilters();
     }
 
@@ -1267,6 +1268,11 @@ public void updateBrokerDispatchRate() {
             dispatchRateInBroker(brokerService.pulsar().getConfiguration()));
     }
 
+    public void updateDispatchPauseOnAckStatePersistentEnabled() {
+        topicPolicies.getDispatcherPauseOnAckStatePersistentEnabled().updateBrokerValue(
+                brokerService.pulsar().getConfiguration().isDispatcherPauseOnAckStatePersistentEnabled());
+    }
+
     public void addFilteredEntriesCount(int filtered) {
         this.filteredEntriesCounter.add(filtered);
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -2605,6 +2605,10 @@ private void updateConfigurationAndRegisterListeners() {
         registerConfigurationListener("dispatchThrottlingRatePerSubscriptionInByte", (dispatchRatePerTopicInByte) -> {
             updateSubscriptionMessageDispatchRate();
         });
+        // add listener to update message-dispatch-rate in byte for subscription
+        registerConfigurationListener("dispatcherPauseOnAckStatePersistentEnabled", (dispatchRatePerTopicInByte) -> {
+            updateDispatchPauseOnAckStatePersistentEnabled();
+        });
 
         // add listener to update message-dispatch-rate in msg for replicator
         registerConfigurationListener("dispatchThrottlingRatePerReplicatorInMsg",
@@ -2743,6 +2747,28 @@ private void updateTopicMessageDispatchRate() {
         });
     }
 
+    private void updateDispatchPauseOnAckStatePersistentEnabled() {
+        this.pulsar().getExecutor().execute(() -> {
+            forEachTopic(topic -> {
+                if (topic instanceof PersistentTopic) {
+                    // Update policies.
+                    PersistentTopic persistentTopic = (PersistentTopic) topic;
+                    persistentTopic.updateDispatchPauseOnAckStatePersistentEnabled();
+                    // Trigger new read if subscriptions has been paused before.
+                    if (!pulsar().getConfiguration().isDispatcherPauseOnAckStatePersistentEnabled()) {
+                        persistentTopic.updateDispatchPauseOnAckStatePersistentEnabled();
+                        persistentTopic.getSubscriptions().forEach((sName, subscription) -> {
+                            if (subscription.getDispatcher() == null) {
+                                return;
+                            }
+                            subscription.getDispatcher().afterAckMessages(null, 0);
+                        });
+                    }
+                }
+            });
+        });
+    }
+
     private void updateBrokerSubscriptionTypesEnabled(Object subscriptionTypesEnabled) {
         this.pulsar().getExecutor().execute(() -> {
             // update subscriptionTypesEnabled

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

@@ -1039,23 +1039,31 @@ public void addUnAckedMessages(int numberOfMessages) {
 
     @Override
     public void afterAckMessages(Throwable exOfDeletion, Object ctxOfDeletion) {
-        if (blockedDispatcherOnCursorDataCanNotFullyPersist == TRUE) {
-            if (cursor.isCursorDataFullyPersistable()) {
-                // If there was no previous pause due to cursor data is too large to persist, we don't need to manually
-                // trigger a new read. This can avoid too many CPU circles.
-                if (BLOCKED_DISPATCHER_ON_CURSOR_DATA_CAN_NOT_FULLY_PERSIST_UPDATER.compareAndSet(this, TRUE, FALSE)) {
-                    readMoreEntriesAsync();
-                } else {
-                    // Retry due to conflict update.
-                    afterAckMessages(exOfDeletion, ctxOfDeletion);
-                }
+        boolean paused = blockedDispatcherOnCursorDataCanNotFullyPersist == TRUE;
+        boolean shouldPauseNow = !cursor.isCursorDataFullyPersistable()
+                && topic.isDispatcherPauseOnAckStatePersistentEnabled();
+        // No need to change.
+        if (paused == shouldPauseNow) {
+            return;
+        }
+        // Should change to "un-pause".
+        if (paused && !shouldPauseNow) {
+            // If there was no previous pause due to cursor data is too large to persist, we don't need to manually
+            // trigger a new read. This can avoid too many CPU circles.
+            if (BLOCKED_DISPATCHER_ON_CURSOR_DATA_CAN_NOT_FULLY_PERSIST_UPDATER.compareAndSet(this, TRUE, FALSE)) {
+                readMoreEntriesAsync();
+            } else {
+                // Retry due to conflict update.
+                afterAckMessages(exOfDeletion, ctxOfDeletion);
             }
-        } else {
-            if (!cursor.isCursorDataFullyPersistable()) {
-                if (BLOCKED_DISPATCHER_ON_CURSOR_DATA_CAN_NOT_FULLY_PERSIST_UPDATER.compareAndSet(this, FALSE, TRUE)) {
-                    // Retry due to conflict update.
-                    afterAckMessages(exOfDeletion, ctxOfDeletion);
-                }
+            return;
+        }
+        // Should change to "paused".
+        if (!paused && shouldPauseNow) {
+            if (!BLOCKED_DISPATCHER_ON_CURSOR_DATA_CAN_NOT_FULLY_PERSIST_UPDATER
+                    .compareAndSet(this, FALSE, TRUE)) {
+                // Retry due to conflict update.
+                afterAckMessages(exOfDeletion, ctxOfDeletion);
             }
         }
     }

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);
 }

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

@@ -37,7 +37,9 @@
 import org.apache.pulsar.client.admin.GetStatsOptions;
 import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.HierarchyTopicPolicies;
 import org.apache.pulsar.common.policies.data.TopicPolicies;
+import org.awaitility.Awaitility;
 import org.awaitility.reflect.WhiteboxImpl;
 import org.testng.Assert;
 import org.testng.annotations.AfterClass;
@@ -211,6 +213,69 @@ public boolean hasAckedMessage(String v) {
         }
     }
 
+    @Test
+    public void testBrokerDynamicConfig() throws Exception {
+        final String tpName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+        final String subscription = "s1";
+        final int msgSendCount = MAX_UNACKED_RANGES_TO_PERSIST * 4;
+        final int incomingQueueSize = MAX_UNACKED_RANGES_TO_PERSIST * 10;
+
+        // Enable "dispatcherPauseOnAckStatePersistentEnabled".
+        admin.brokers().updateDynamicConfiguration("dispatcherPauseOnAckStatePersistentEnabled", "true");
+        admin.topics().createNonPartitionedTopic(tpName);
+        admin.topics().createSubscription(tpName, subscription, MessageId.earliest);
+
+        PersistentTopic persistentTopic =
+                (PersistentTopic) pulsar.getBrokerService().getTopic(tpName, false).join().get();
+        Awaitility.await().untilAsserted(() -> {
+            Assert.assertTrue(pulsar.getConfig().isDispatcherPauseOnAckStatePersistentEnabled());
+            HierarchyTopicPolicies policies = WhiteboxImpl.getInternalState(persistentTopic, "topicPolicies");
+            Boolean v = policies.getDispatcherPauseOnAckStatePersistentEnabled().get();
+            Assert.assertNotNull(v);
+            Assert.assertTrue(v.booleanValue());
+        });
+
+        // Send double MAX_UNACKED_RANGES_TO_PERSIST messages.
+        Producer<String> p1 = pulsarClient.newProducer(Schema.STRING).topic(tpName).enableBatching(false).create();
+        ArrayList<MessageId> messageIdsSent = new ArrayList<>();
+        for (int i = 0; i < msgSendCount; i++) {
+            MessageIdImpl messageId = (MessageIdImpl) p1.send(Integer.valueOf(i).toString());
+            messageIdsSent.add(messageId);
+        }
+        // Make ack holes.
+        Consumer<String> c1 = pulsarClient.newConsumer(Schema.STRING).topic(tpName).subscriptionName(subscription)
+                .receiverQueueSize(incomingQueueSize).isAckReceiptEnabled(true)
+                .subscriptionType(SubscriptionType.Shared).subscribe();
+        ackOddMessagesOnly(c1);
+
+        cancelPendingRead(tpName, subscription);
+        triggerNewReadMoreEntries(tpName, subscription);
+
+        // Verify: the dispatcher has been paused.
+        final String specifiedMessage = "9876543210";
+        p1.send(specifiedMessage);
+        Message<String> msg1 = c1.receive(2, TimeUnit.SECONDS);
+        Assert.assertNull(msg1, msg1 == null ? "null" : msg1.getValue());
+
+        // Disable "dispatcherPauseOnAckStatePersistentEnabled".
+        admin.brokers().updateDynamicConfiguration("dispatcherPauseOnAckStatePersistentEnabled", "false");
+        Awaitility.await().untilAsserted(() -> {
+            Assert.assertFalse(pulsar.getConfig().isDispatcherPauseOnAckStatePersistentEnabled());
+            HierarchyTopicPolicies policies = WhiteboxImpl.getInternalState(persistentTopic, "topicPolicies");
+            Boolean v = policies.getDispatcherPauseOnAckStatePersistentEnabled().get();
+            Assert.assertTrue(v == null || !v.booleanValue());
+        });
+
+        // Verify the new message can be received.
+        Message<String> msg2 = c1.receive(2, TimeUnit.SECONDS);
+        Assert.assertNotNull(msg2);
+        Assert.assertEquals(msg2.getValue(), specifiedMessage);
+        // cleanup.
+        p1.close();
+        c1.close();
+        admin.topics().delete(tpName, false);
+    }
+
     @Test(dataProvider = "multiConsumerSubscriptionTypes")
     public void testPauseOnAckStatPersist(SubscriptionType subscriptionType) throws Exception {
         final String tpName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");

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<byte[]> 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<byte[]> 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<CommandWatchTopicListSuccess> newWatchTopicList(
+                        BaseCommand command, long requestId) {
+                    // Inject 2 seconds delay when sending command New Watch Topics.
+                    CompletableFuture<CommandWatchTopicListSuccess> 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.