(iotdb) branch dev/1.3 updated: Subscription: check if all topic messages have been consumed when client polling (#14937) (#14953)

[rong]

This is an automated email from the ASF dual-hosted git repository.

rong pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/dev/1.3 by this push:
     new 5680d810668 Subscription: check if all topic messages have been 
consumed when client polling (#14937) (#14953)
5680d810668 is described below

commit 5680d8106681c3ef7e2f4864a33b72aa69af30c1
Author: VGalaxies <vgalax...@apache.org>
AuthorDate: Thu Feb 27 11:09:54 2025 +0800

    Subscription: check if all topic messages have been consumed when client 
polling (#14937) (#14953)
---
 .../consumer/SubscriptionConsumer.java             | 22 +++++++++++++++++-----
 1 file changed, 17 insertions(+), 5 deletions(-)

diff --git 
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java
 
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java
index d9d4bcd2cf1..d760aa2b3be 100644
--- 
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java
+++ 
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java
@@ -63,6 +63,7 @@ import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -83,7 +84,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 
-import static 
org.apache.iotdb.rpc.subscription.config.TopicConstant.MODE_SNAPSHOT_VALUE;
 import static 
org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseType.ERROR;
 import static 
org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseType.FILE_INIT;
 import static 
org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseType.TABLETS;
@@ -124,9 +124,16 @@ abstract class SubscriptionConsumer implements 
AutoCloseable {
   protected volatile Map<String, TopicConfig> subscribedTopics = new 
HashMap<>();
 
   public boolean allSnapshotTopicMessagesHaveBeenConsumed() {
-    return subscribedTopics.values().stream()
-        .noneMatch(
-            (config) -> 
config.getAttributesWithSourceMode().containsValue(MODE_SNAPSHOT_VALUE));
+    return allTopicMessagesHaveBeenConsumed(subscribedTopics.keySet());
+  }
+
+  private boolean allTopicMessagesHaveBeenConsumed(final Collection<String> 
topicNames) {
+    // For the topic that needs to be detected, there are two scenarios to 
consider:
+    //   1. If configs as live, it cannot be determined whether the topic has 
been fully consumed.
+    //   2. If configs as snapshot, it means the topic has not been 
automatically unsubscribed.
+    // Therefore, the logic can be summarized as follows: if there is a 
matching topic in subscribed
+    // topics, then it has not been fully consumed.
+    return 
topicNames.stream().map(subscribedTopics::get).noneMatch(Objects::nonNull);
   }
 
   /////////////////////////////// getter ///////////////////////////////
@@ -486,7 +493,7 @@ abstract class SubscriptionConsumer implements 
AutoCloseable {
                         final String topicNameToUnsubscribe = 
commitContext.getTopicName();
                         LOGGER.info(
                             "Termination occurred when SubscriptionConsumer {} 
polling topics, unsubscribe topic {} automatically",
-                            this,
+                            coreReportMessage(),
                             topicNameToUnsubscribe);
                         
unsubscribe(Collections.singleton(topicNameToUnsubscribe), false);
                         return Optional.empty();
@@ -660,6 +667,11 @@ abstract class SubscriptionConsumer implements 
AutoCloseable {
           break;
         }
 
+        // check if all topic messages have been consumed
+        if (allTopicMessagesHaveBeenConsumed(topicNames)) {
+          break;
+        }
+
         // update timer
         timer.update();