Subscription: adapt pipe completion signal for automatically drop subscription

example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java

@@ -25,16 +25,21 @@
 import org.apache.iotdb.rpc.subscription.config.TopicConstant;
 import org.apache.iotdb.session.Session;
 import org.apache.iotdb.session.subscription.SubscriptionSession;
+import org.apache.iotdb.session.subscription.consumer.AckStrategy;
+import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
 import org.apache.iotdb.session.subscription.consumer.SubscriptionPullConsumer;
+import org.apache.iotdb.session.subscription.consumer.SubscriptionPushConsumer;
 import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
 import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionTsFileHandler;
 
 import org.apache.tsfile.read.TsFileReader;
 import org.apache.tsfile.read.common.Path;
 import org.apache.tsfile.read.expression.QueryExpression;
 import org.apache.tsfile.read.query.dataset.QueryDataSet;
 
 import java.io.IOException;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -51,6 +56,8 @@ public class SubscriptionSessionExample {
 
   private static final String TOPIC_1 = "topic1";
   private static final String TOPIC_2 = "`'topic2'`";
+  private static final String TOPIC_3 = "`\"topic3\"`";
+  private static final String TOPIC_4 = "`\"top \\.i.c4\"`";
 
   private static final long SLEEP_NS = 1_000_000_000L;
   private static final long POLL_TIMEOUT_MS = 10_000L;
@@ -226,10 +233,109 @@ private static void dataSubscription2() throws Exception {
     }
   }
 
+  /** multi push consumer subscribe topic with tsfile format and query mode */
+  private static void dataSubscription3() throws Exception {
+    try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) {
+      subscriptionSession.open();
+      final Properties config = new Properties();
+      config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE);
+      config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_QUERY_VALUE);
+      subscriptionSession.createTopic(TOPIC_3, config);
+    }
+
+    final List<Thread> threads = new ArrayList<>();
+    for (int i = 0; i < PARALLELISM; ++i) {
+      final int idx = i;
+      final Thread thread =
+          new Thread(
+              () -> {
+                // Subscription: builder-style ctor
+                try (final SubscriptionPushConsumer consumer3 =
+                    new SubscriptionPushConsumer.Builder()
+                        .consumerId("c" + idx)
+                        .consumerGroupId("cg3")
+                        .ackStrategy(AckStrategy.AFTER_CONSUME)
+                        .consumeListener(
+                            message -> {
+                              // do something for SubscriptionTsFileHandler
+                              System.out.println(
+                                  message.getTsFileHandler().getFile().getAbsolutePath());
+                              return ConsumeResult.SUCCESS;
+                            })
+                        .buildPushConsumer()) {
+                  consumer3.open();
+                  consumer3.subscribe(TOPIC_3);
+                  while (consumer3.hasMoreData()) {
+                    LockSupport.parkNanos(SLEEP_NS); // wait some time
+                  }
+                }
+              });
+      thread.start();
+      threads.add(thread);
+    }
+
+    for (final Thread thread : threads) {
+      thread.join();
+    }
+  }
+
+  /** multi pull consumer subscribe topic with tsfile format and query mode */
+  private static void dataSubscription4() throws Exception {
+    try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) {
+      subscriptionSession.open();
+      final Properties config = new Properties();
+      config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE);
+      config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_QUERY_VALUE);
+      subscriptionSession.createTopic(TOPIC_4, config);
+    }
+
+    final List<Thread> threads = new ArrayList<>();
+    for (int i = 0; i < PARALLELISM; ++i) {
+      final int idx = i;
+      final Thread thread =
+          new Thread(
+              () -> {
+                // Subscription: builder-style ctor
+                try (final SubscriptionPullConsumer consumer4 =
+                    new SubscriptionPullConsumer.Builder()
+                        .consumerId("c" + idx)
+                        .consumerGroupId("cg4")
+                        .autoCommit(true)
+                        .fileSaveFsync(true)
+                        .buildPullConsumer()) {
+                  consumer4.open();
+                  consumer4.subscribe(TOPIC_4);
+                  while (true) {
+                    LockSupport.parkNanos(SLEEP_NS); // wait some time
+                    if (!consumer4.hasMoreData()) {
+                      break;
+                    }
+                    for (final SubscriptionMessage message : consumer4.poll(POLL_TIMEOUT_MS)) {
+                      final SubscriptionTsFileHandler handler = message.getTsFileHandler();
+                      handler.moveFile(
+                          Paths.get(System.getProperty("user.dir"), "new-subscription-dir")
+                              .resolve(handler.getPath().getFileName()));
+                    }
+                  }
+                } catch (final IOException e) {
+                  throw new RuntimeException(e);
+                }
+              });
+      thread.start();
+      threads.add(thread);
+    }
+
+    for (final Thread thread : threads) {
+      thread.join();
+    }
+  }
+
   public static void main(final String[] args) throws Exception {
     prepareData();
     // dataQuery();
     // dataSubscription1();
-    dataSubscription2();
+    // dataSubscription2();
+    // dataSubscription3();
+    dataSubscription4();
   }
 }

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java

@@ -438,6 +438,32 @@ public CommonConfig setCnConnectionTimeoutMs(int connectionTimeoutMs) {
     return this;
   }
 
+  @Override
+  public CommonConfig setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(
+      int pipeHeartbeatIntervalSecondsForCollectingPipeMeta) {
+    setProperty(
+        "pipe_heartbeat_interval_seconds_for_collecting_pipe_meta",
+        String.valueOf(pipeHeartbeatIntervalSecondsForCollectingPipeMeta));
+    return this;
+  }
+
+  @Override
+  public CommonConfig setPipeMetaSyncerInitialSyncDelayMinutes(
+      long pipeMetaSyncerInitialSyncDelayMinutes) {
+    setProperty(
+        "pipe_meta_syncer_initial_sync_delay_minutes",
+        String.valueOf(pipeMetaSyncerInitialSyncDelayMinutes));
+    return this;
+  }
+
+  @Override
+  public CommonConfig setPipeMetaSyncerSyncIntervalMinutes(long pipeMetaSyncerSyncIntervalMinutes) {
+    setProperty(
+        "pipe_meta_syncer_sync_interval_minutes",
+        String.valueOf(pipeMetaSyncerSyncIntervalMinutes));
+    return this;
+  }
+
   // For part of the log directory
   public String getClusterConfigStr() {
     return fromConsensusFullNameToAbbr(properties.getProperty(CONFIG_NODE_CONSENSUS_PROTOCOL_CLASS))

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java

@@ -445,4 +445,29 @@ public CommonConfig setCnConnectionTimeoutMs(int connectionTimeoutMs) {
     cnConfig.setCnConnectionTimeoutMs(connectionTimeoutMs);
     return this;
   }
+
+  @Override
+  public CommonConfig setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(
+      int pipeHeartbeatIntervalSecondsForCollectingPipeMeta) {
+    dnConfig.setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(
+        pipeHeartbeatIntervalSecondsForCollectingPipeMeta);
+    cnConfig.setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(
+        pipeHeartbeatIntervalSecondsForCollectingPipeMeta);
+    return this;
+  }
+
+  @Override
+  public CommonConfig setPipeMetaSyncerInitialSyncDelayMinutes(
+      long pipeMetaSyncerInitialSyncDelayMinutes) {
+    dnConfig.setPipeMetaSyncerInitialSyncDelayMinutes(pipeMetaSyncerInitialSyncDelayMinutes);
+    cnConfig.setPipeMetaSyncerInitialSyncDelayMinutes(pipeMetaSyncerInitialSyncDelayMinutes);
+    return this;
+  }
+
+  @Override
+  public CommonConfig setPipeMetaSyncerSyncIntervalMinutes(long pipeMetaSyncerSyncIntervalMinutes) {
+    dnConfig.setPipeMetaSyncerSyncIntervalMinutes(pipeMetaSyncerSyncIntervalMinutes);
+    cnConfig.setPipeMetaSyncerSyncIntervalMinutes(pipeMetaSyncerSyncIntervalMinutes);
+    return this;
+  }
 }

integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java

@@ -313,4 +313,21 @@ public CommonConfig setTagAttributeTotalSize(int tagAttributeTotalSize) {
   public CommonConfig setCnConnectionTimeoutMs(int connectionTimeoutMs) {
     return this;
   }
+
+  @Override
+  public CommonConfig setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(
+      int pipeHeartbeatIntervalSecondsForCollectingPipeMeta) {
+    return this;
+  }
+
+  @Override
+  public CommonConfig setPipeMetaSyncerInitialSyncDelayMinutes(
+      long pipeMetaSyncerInitialSyncDelayMinutes) {
+    return this;
+  }
+
+  @Override
+  public CommonConfig setPipeMetaSyncerSyncIntervalMinutes(long pipeMetaSyncerSyncIntervalMinutes) {
+    return this;
+  }
 }

integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java

@@ -139,4 +139,11 @@ CommonConfig setEnableAutoLeaderBalanceForIoTConsensus(
   CommonConfig setTagAttributeTotalSize(int tagAttributeTotalSize);
 
   CommonConfig setCnConnectionTimeoutMs(int connectionTimeoutMs);
+
+  CommonConfig setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(
+      int pipeHeartbeatIntervalSecondsForCollectingPipeMeta);
+
+  CommonConfig setPipeMetaSyncerInitialSyncDelayMinutes(long pipeMetaSyncerInitialSyncDelayMinutes);
+
+  CommonConfig setPipeMetaSyncerSyncIntervalMinutes(long pipeMetaSyncerSyncIntervalMinutes);
 }