@Test
  public void do_test()
      throws InterruptedException,
          TException,
          IoTDBConnectionException,
          IOException,
          StatementExecutionException {
    // Write data before subscribing
    insert_data(1706659200000L); // 2024-01-31 08:00:00+08:00
    session_src.executeNonQueryStatement(
        "insert into "
            + device
            + "(time,s_0,s_1)values(1710288000000,313,'2024-03-13 08:00:00+08:00');"); // 2024-03-13
    // 08:00:00+08:00
    session_src.executeNonQueryStatement("flush");
    // Subscribe
    consumer = create_pull_consumer("multi_1consumer_mix", "tsfile_dataset", false, null);
    System.out.println("###### Before Subscription Query:");
    subs.getSubscriptions().forEach((System.out::println));
    assertEquals(subs.getSubscriptions().size(), 0, "Before subscription show subscriptions");
    consumer.subscribe(topicName, topicName2);
    long timestamp = System.currentTimeMillis();
    System.out.println("###### Subscription Query:");
    subs.getSubscriptions().forEach((System.out::println));
    assertEquals(subs.getSubscriptions().size(), 2, "show subscriptions after subscription");
    // Subscribe and then write data
    Thread thread =
        new Thread(
            new Runnable() {
              @Override
              public void run() {
                try {
                  insert_data(System.currentTimeMillis());
                } catch (Exception e) {
                  throw new RuntimeException(e);
                }
              }
            });
    thread.start();

    AtomicInteger rowCount = new AtomicInteger(0);
    Thread thread2 =
        new Thread(
            () -> {
              while (true) {
                List<SubscriptionMessage> messages = consumer.poll(Duration.ofMillis(10000));
                if (messages.isEmpty()) {
                  break;
                }

                for (final SubscriptionMessage message : messages) {
                  final short messageType = message.getMessageType();
                  if (SubscriptionMessageType.isValidatedMessageType(messageType)) {
                    switch (SubscriptionMessageType.valueOf(messageType)) {
                      case SESSION_DATA_SETS_HANDLER:
                        for (final Iterator<Tablet> it =
                                message.getSessionDataSetsHandler().tabletIterator();
                            it.hasNext(); ) {
                          final Tablet tablet = it.next();
                          try {
                            session_dest.insertTablet(tablet);
                            //
                            // System.out.println(format.format(new Date())+" "+tablet.rowSize);
                          } catch (StatementExecutionException e) {
                            throw new RuntimeException(e);
                          } catch (IoTDBConnectionException e) {
                            throw new RuntimeException(e);
                          }
                        }
                        break;
                      case TS_FILE_HANDLER:
                        try {
                          TsFileReader reader = message.getTsFileHandler().openReader();
                          QueryDataSet dataset =
                              reader.query(
                                  QueryExpression.create(
                                      Collections.singletonList(new Path(device, "s_1", true)),
                                      null));
                          while (dataset.hasNext()) {
                            rowCount.addAndGet(1);
                            RowRecord next = dataset.next();
                            System.out.println(
                                device + ".s_1:" + next.getTimestamp() + "," + next.getFields());
                          }
                        } catch (IOException e) {
                          throw new RuntimeException(e);
                        }
                        consumer.commitSync(messages);
                        break;
                    }
                  }
                }
              }
            });
    Thread thread3 =
        new Thread(
            () -> {
              while (true) {
                List<SubscriptionMessage> messages = consumer.poll(Duration.ofMillis(10000));
                if (messages.isEmpty()) {
                  break;
                }

                for (final SubscriptionMessage message : messages) {
                  final short messageType = message.getMessageType();
                  if (SubscriptionMessageType.isValidatedMessageType(messageType)) {
                    switch (SubscriptionMessageType.valueOf(messageType)) {
                      case SESSION_DATA_SETS_HANDLER:
                        for (final Iterator<Tablet> it =
                                message.getSessionDataSetsHandler().tabletIterator();
                            it.hasNext(); ) {
                          final Tablet tablet = it.next();
                          try {
                            session_dest.insertTablet(tablet);
                            System.out.println(
                                FORMAT.format(new Date()) + " " + tablet.getRowSize());
                          } catch (StatementExecutionException e) {
                            throw new RuntimeException(e);
                          } catch (IoTDBConnectionException e) {
                            throw new RuntimeException(e);
                          }
                        }
                        break;
                      case TS_FILE_HANDLER:
                        try {
                          TsFileReader reader = message.getTsFileHandler().openReader();
                          QueryDataSet dataset =
                              reader.query(
                                  QueryExpression.create(
                                      Collections.singletonList(new Path(device, "s_1", true)),
                                      null));
                          while (dataset.hasNext()) {
                            rowCount.addAndGet(1);
                            RowRecord next = dataset.next();
                            System.out.println(
                                device + ":" + next.getTimestamp() + "," + next.getFields());
                          }
                        } catch (IOException e) {
                          throw new RuntimeException(e);
                        }
                        consumer.commitSync(messages);
                        break;
                    }
                  }
                }
              }
            });
    thread2.start();
    thread.join();
    thread2.join(5000);
    String sql1 = "select count(s_0) from " + device + " where time <= " + timestamp;
    String sql2 = "select count(s_1) from " + device + " where time <= " + timestamp;

    System.out.println(FORMAT.format(new Date()) + " src:" + getCount(session_src, sql1));
    AWAIT.untilAsserted(
        () -> {
          // Consumption data
          check_count(6, sql1, "Consumption data:" + pattern);
          check_count(0, sql2, "Consumption data:" + pattern2);
          assertEquals(rowCount.get(), 6, "tsfile consumer");
        });
    // Unsubscribe
    consumer.unsubscribe(topicName);
    System.out.println("###### After unsubscribing query:");
    subs.getSubscriptions().forEach((System.out::println));
    assertEquals(subs.getSubscriptions().size(), 1, "Unsubscribe 1 and then show subscriptions");

    // Unsubscribe and then write data
    insert_data(1707782400000L); // 2024-02-13 08:00:00+08:00
    session_src.executeNonQueryStatement(
        "insert into "
            + device
            + "(time,s_0,s_1)values(1703980800000,3.45,'2023-12-31 08:00:00+08:00');"); // 2023-12-31 08:00:00+08:00
    insert_data(System.currentTimeMillis());
    session_src.executeNonQueryStatement("flush");
    System.out.println(
        FORMAT.format(new Date())
            + " Unsubscribe after writing data src:"
            + getCount(session_src, sql1));

    thread3.start();
    thread3.join(5000);
    System.out.println(FORMAT.format(new Date()));
    AWAIT.untilAsserted(
        () -> {
          assertEquals(
              rowCount.get(), 12, "Re-consume data: tsfile consumer " + FORMAT.format(new Date()));
          check_count(6, sql1, "consume data again:" + pattern);
          check_count(0, sql2, "Reconsume data:" + pattern2);
        });
    // close
    consumer.close();
    try {
      consumer.subscribe(topicName, topicName2);
    } catch (Exception e) {
      System.out.println("subscribe again after close, expecting an exception");
    }
    assertEquals(subs.getSubscriptions().size(), 0, "show subscriptions after close");
    subs.getSubscriptions().forEach((System.out::println));
  }
1 consumer subscribes to 2 topics: Historical data
integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/multi/IoTDBOneConsumerMultiTopicsMixIT.java