@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
    System.out.println("### TestPushConsumeNoTargetDirTsfile ###");

    final AtomicInteger onReceiveCount = new AtomicInteger(0);
    final AtomicInteger rowCount = new AtomicInteger(0);
    consumer =
        new SubscriptionTreePushConsumer.Builder()
            .host(SRC_HOST)
            .port(SRC_PORT)
            .ackStrategy(AckStrategy.BEFORE_CONSUME)
            .consumeListener(
                message -> {
                  onReceiveCount.incrementAndGet();
                  try {
                    TsFileReader reader = message.getTsFileHandler().openReader();
                    Path path = new Path(device, "s_0", true);
                    QueryDataSet dataset =
                        reader.query(QueryExpression.create(Collections.singletonList(path), null));
                    while (dataset.hasNext()) {
                      rowCount.addAndGet(1);
                      dataset.next();
                    }
                  } catch (IOException e) {
                    throw new RuntimeException(e);
                  }
                  return ConsumeResult.SUCCESS;
                })
            .buildPushConsumer();
    consumer.open();
    // Subscribe
    consumer.subscribe(topicName);
    subs.getSubscriptions().forEach(System.out::println);
    assertEquals(subs.getSubscriptions().size(), 1, "show subscriptions after subscription");
    insert_data(System.currentTimeMillis());

    AWAIT.untilAsserted(
        () -> {
          assertEquals(onReceiveCount.get(), 2, "should 2 tsfile");
          assertEquals(rowCount.get(), 10, "should process 8 rows data");
        });

    // Unsubscribe
    consumer.unsubscribe(topicName);
    assertEquals(subs.getSubscriptions().size(), 0, "show subscriptions after unsubscription");
    // Subscribe and then write data
    consumer.subscribe(topicName);
    assertEquals(subs.getSubscriptions().size(), 1, "show subscriptions after re-subscribing");
    insert_data(1707782400000L); // 2024-02-13 08:00:00+08:00
    AWAIT.untilAsserted(
        () -> {
          // Consumption data: Progress is not retained after canceling and re-subscribing. Full
          // synchronization.
          assertEquals(onReceiveCount.get(), 5, "should 5 tsfiles include 2 duplicates");
          assertEquals(rowCount.get(), 25, "should process 20 rows data, include ");
        });
  }
PushConsumer:BEFORE_CONSUME
TsFileHandler
pattern: db
target: no
consumer_id: no
group_id: no
integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeNoTargetDirTsfileIT.java