@Test
  public void do_test()
      throws InterruptedException,
          TException,
          IoTDBConnectionException,
          IOException,
          StatementExecutionException {
    // Write data before subscribing
    insert_data(1706659200000L, device); // 2024-01-31 08:00:00+08:00
    insert_data(1706659200000L, device2); // 2024-01-31 08:00:00+08:00

    consumer =
        new SubscriptionTreePullConsumer.Builder()
            .host(SRC_HOST)
            .port(SRC_PORT)
            .consumerId("snapshot_ts_pattern_tsfile_pull")
            .consumerGroupId("loose_range_all")
            .autoCommit(false)
            .fileSaveDir("target/pull-subscription") // hack for license check
            .buildPullConsumer();
    consumer.open();
    // Subscribe
    consumer.subscribe(topicName);
    subs.getSubscriptions().forEach(System.out::println);
    assertEquals(subs.getSubscriptions().size(), 1, "show subscriptions after subscription");

    insert_data(nowTimestamp - 4000, device);
    insert_data(nowTimestamp - 4000, device2);
    System.out.println(
        FORMAT.format(new Date())
            + " src filter:"
            + getCount(
                session_src, "select count(s_0) from " + device + " where time <=" + nowTimestamp));

    // Consumption data
    List<String> paths = new ArrayList<>(3);
    paths.add(device);
    paths.add(device2);
    paths.add(database2 + ".d_2");
    List<Integer> rowCountList = consume_tsfile(consumer, paths);
    // Subscribe and write without consuming
    assertEquals(rowCountList.get(0), 5, "Write without consume after subscription");
    assertEquals(rowCountList.get(1), 0);
    assertEquals(rowCountList.get(2), 0);

    // Unsubscribe
    consumer.unsubscribe(topicName);
    assertEquals(subs.getSubscriptions().size(), 0, "After cancellation, show subscriptions");

    // Subscribe and then write data
    consumer.subscribe(topicName);
    assertEquals(subs.getSubscriptions().size(), 1, "show subscriptions after re-subscribing");
    insert_data(1707782400000L, device); // 2024-02-13 08:00:00+08:00
    insert_data(1707782400000L, device2); // 2024-02-13 08:00:00+08:00

    // Consumption data: Progress is not retained after unsubscribing and re-subscribing. Full
    // synchronization.
    rowCountList = consume_tsfile(consumer, paths);
    assertEquals(
        rowCountList.get(0),
        10,
        "Unsubscribe and then resubscribe, progress is not retained. Full synchronization.");
    assertEquals(rowCountList.get(1), 0, "Unsubscribe and then resubscribe," + device2);
    assertEquals(rowCountList.get(2), 0, "Unsubscribe and then resubscribe," + database2 + ".d_2");
  }
PullConsumer
pattern: ts
format: tsfile
loose-range: all
mode: snapshot
integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBAllTsfilePullConsumerSnapshotIT.java