Mergemaster0620

example/pipe-count-point-processor/src/main/java/org/apache/iotdb/CountPointProcessor.java

@@ -44,26 +44,28 @@ public class CountPointProcessor implements PipeProcessor {
   private PartialPath aggregateSeries;
 
   @Override
-  public void validate(PipeParameterValidator validator) {
+  public void validate(final PipeParameterValidator validator) {
     validator.validateRequiredAttribute(AGGREGATE_SERIES_KEY);
   }
 
   @Override
-  public void customize(PipeParameters parameters, PipeProcessorRuntimeConfiguration configuration)
+  public void customize(
+      final PipeParameters parameters, final PipeProcessorRuntimeConfiguration configuration)
       throws Exception {
     this.aggregateSeries = new PartialPath(parameters.getString(AGGREGATE_SERIES_KEY));
   }
 
   @Override
-  public void process(TabletInsertionEvent tabletInsertionEvent, EventCollector eventCollector) {
+  public void process(
+      final TabletInsertionEvent tabletInsertionEvent, final EventCollector eventCollector) {
     tabletInsertionEvent.processTablet(
         (tablet, rowCollector) -> writePointCount.addAndGet(tablet.rowSize));
   }
 
   @Override
-  public void process(Event event, EventCollector eventCollector) throws Exception {
+  public void process(final Event event, final EventCollector eventCollector) throws Exception {
     if (event instanceof PipeHeartbeatEvent) {
-      Tablet tablet =
+      final Tablet tablet =
           new Tablet(
               aggregateSeries.getIDeviceID().toString(),
               Collections.singletonList(
@@ -73,7 +75,7 @@ public void process(Event event, EventCollector eventCollector) throws Exception
       tablet.addTimestamp(0, System.currentTimeMillis());
       tablet.addValue(aggregateSeries.getMeasurement(), 0, writePointCount.get());
       eventCollector.collect(
-          new PipeRawTabletInsertionEvent(tablet, false, null, null, null, false));
+          new PipeRawTabletInsertionEvent(tablet, false, null, 0, null, null, false));
     }
   }
 

example/schema/pom.xml

@@ -32,7 +32,6 @@
         <dependency>
             <groupId>ch.qos.logback</groupId>
             <artifactId>logback-classic</artifactId>
-            <version>1.3.13</version>
         </dependency>
         <dependency>
             <groupId>org.apache.tsfile</groupId>

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

@@ -25,14 +25,23 @@
 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;
 import java.util.List;
 import java.util.Properties;
@@ -46,11 +55,14 @@ public class SubscriptionSessionExample {
   private static final int PORT = 6667;
 
   private static final String TOPIC_1 = "topic1";
-  private static final String TOPIC_2 = "`topic2`";
+  private static final String TOPIC_2 = "`'topic2'`";
+  private static final String TOPIC_3 = "`\"topic3\"`";
+  private static final String TOPIC_4 = "`\"top \\.i.c4\"`";
 
-  public static final long SLEEP_NS = 1_000_000_000L;
-  public static final long POLL_TIMEOUT_MS = 10_000L;
+  private static final long SLEEP_NS = 1_000_000_000L;
+  private static final long POLL_TIMEOUT_MS = 10_000L;
   private static final int MAX_RETRY_TIMES = 3;
+  private static final int PARALLELISM = 8;
 
   private static void prepareData() throws Exception {
     // Open session
@@ -66,13 +78,13 @@ private static void prepareData() throws Exception {
 
     // Insert some historical data
     final long currentTime = System.currentTimeMillis();
-    for (int i = 0; i < 10000; ++i) {
+    for (int i = 0; i < 100; ++i) {
       session.executeNonQueryStatement(
           String.format("insert into root.db.d1(time, s1, s2) values (%s, 1, 2)", i));
       session.executeNonQueryStatement(
-          String.format("insert into root.db.d2(time, s3, s4) values (%s, 3, 4)", currentTime + i));
+          String.format("insert into root.db.d2(time, s1, s2) values (%s, 3, 4)", currentTime + i));
       session.executeNonQueryStatement(
-          String.format("insert into root.sg.d3(time, s5) values (%s, 5)", currentTime + 2 * i));
+          String.format("insert into root.sg.d3(time, s1) values (%s, 5)", currentTime + 2 * i));
     }
     session.executeNonQueryStatement("flush");
 
@@ -104,11 +116,16 @@ private static void dataQuery() throws Exception {
     session = null;
   }
 
-  private static void subscriptionExample1() throws Exception {
+  /** single pull consumer subscribe topic with path and time range */
+  private static void dataSubscription1() throws Exception {
     // Create topics
     try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) {
       subscriptionSession.open();
-      subscriptionSession.createTopic(TOPIC_1);
+      final Properties config = new Properties();
+      config.put(TopicConstant.PATH_KEY, "root.db.d1.s1");
+      config.put(TopicConstant.START_TIME_KEY, 25);
+      config.put(TopicConstant.END_TIME_KEY, 75);
+      subscriptionSession.createTopic(TOPIC_1, config);
     }
 
     int retryCount = 0;
@@ -151,7 +168,8 @@ private static void subscriptionExample1() throws Exception {
     consumer1.close();
   }
 
-  private static void subscriptionExample2() throws Exception {
+  /** multi pull consumer subscribe topic with tsfile format */
+  private static void dataSubscription2() throws Exception {
     try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) {
       subscriptionSession.open();
       final Properties config = new Properties();
@@ -160,7 +178,7 @@ private static void subscriptionExample2() throws Exception {
     }
 
     final List<Thread> threads = new ArrayList<>();
-    for (int i = 0; i < 8; ++i) {
+    for (int i = 0; i < PARALLELISM; ++i) {
       final int idx = i;
       final Thread thread =
           new Thread(
@@ -187,7 +205,16 @@ private static void subscriptionExample2() throws Exception {
                     }
                     for (final SubscriptionMessage message : messages) {
                       try (final TsFileReader reader = message.getTsFileHandler().openReader()) {
-                        // do something...
+                        final QueryDataSet dataSet =
+                            reader.query(
+                                QueryExpression.create(
+                                    Arrays.asList(
+                                        new Path("root.db.d2", "s2", true),
+                                        new Path("root.db.d3", "s1", true)),
+                                    null));
+                        while (dataSet.hasNext()) {
+                          System.out.println(dataSet.next());
+                        }
                       }
                     }
                     consumer2.commitSync(messages);
@@ -206,10 +233,109 @@ private static void subscriptionExample2() 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();
-    subscriptionExample1();
-    subscriptionExample2();
+    // dataQuery();
+    // dataSubscription1();
+    // dataSubscription2();
+    // dataSubscription3();
+    dataSubscription4();
   }
 }

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

@@ -120,9 +120,6 @@ public class ClusterConstant {
   public static final String COMMON_PROPERTIES_FILE = "iotdb-common.properties";
   public static final String IOTDB_SYSTEM_PROPERTIES_FILE = "iotdb-system.properties";
 
-  public static final String SYSTEM_PROPERTIES_FILE = "system.properties";
-  public static final String CONFIG_NODE_SYSTEM_PROPERTIES_FILE = "confignode-system.properties";
-
   // Properties' keys
   // Common
   public static final String CONFIG_NODE_CONSENSUS_PROTOCOL_CLASS =

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

@@ -31,8 +31,8 @@
 import java.util.stream.IntStream;
 
 import static org.apache.iotdb.consensus.ConsensusFactory.FAST_IOT_CONSENSUS;
-import static org.apache.iotdb.consensus.ConsensusFactory.IOTV2_CONSENSUS;
 import static org.apache.iotdb.consensus.ConsensusFactory.IOT_CONSENSUS;
+import static org.apache.iotdb.consensus.ConsensusFactory.IOT_CONSENSUS_V2;
 import static org.apache.iotdb.consensus.ConsensusFactory.RATIS_CONSENSUS;
 import static org.apache.iotdb.consensus.ConsensusFactory.REAL_PIPE_CONSENSUS;
 import static org.apache.iotdb.consensus.ConsensusFactory.SIMPLE_CONSENSUS;
@@ -224,7 +224,7 @@ public static String fromConsensusFullNameToAbbr(String consensus) {
         return IOT_CONSENSUS_STR;
       case REAL_PIPE_CONSENSUS:
         return PIPE_CONSENSUS_STR;
-      case IOTV2_CONSENSUS:
+      case IOT_CONSENSUS_V2:
         return STREAM_CONSENSUS_STR;
       case FAST_IOT_CONSENSUS:
         return BATCH_CONSENSUS_STR;
@@ -244,7 +244,7 @@ public static String fromConsensusAbbrToFullName(String consensus) {
       case PIPE_CONSENSUS_STR:
         return REAL_PIPE_CONSENSUS;
       case STREAM_CONSENSUS_STR:
-        return IOTV2_CONSENSUS;
+        return IOT_CONSENSUS_V2;
       case BATCH_CONSENSUS_STR:
         return FAST_IOT_CONSENSUS;
       default:

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/cluster/env/AbstractEnv.java

@@ -640,10 +640,10 @@ public void setTestMethodName(String testMethodName) {
   @Override
   public void dumpTestJVMSnapshot() {
     for (ConfigNodeWrapper configNodeWrapper : configNodeWrapperList) {
-      configNodeWrapper.dumpJVMSnapshot(testMethodName);
+      configNodeWrapper.executeJstack(testMethodName);
     }
     for (DataNodeWrapper dataNodeWrapper : dataNodeWrapperList) {
-      dataNodeWrapper.dumpJVMSnapshot(testMethodName);
+      dataNodeWrapper.executeJstack(testMethodName);
     }
   }