@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("pull_path_device_tsfile")
            .consumerGroupId("loose_range_path")
            .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(System.currentTimeMillis(), device);
    insert_data(System.currentTimeMillis(), device2);
    // Consumption data
    List<String> devices = new ArrayList<>(3);
    devices.add(device);
    devices.add(device2);
    devices.add(database2 + ".d_2");
    List<Integer> results = consume_tsfile(consumer, devices);

    assertEquals(results.get(0), 10);
    assertEquals(results.get(1), 0);
    assertEquals(results.get(2), 0);
    assertEquals(results.get(3), 2, "number of received files");
    // Unsubscribe
    consumer.unsubscribe(topicName);
    assertEquals(subs.getSubscriptions().size(), 0, "show subscriptions after cancellation");
    // 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.
    results = consume_tsfile(consumer, devices);
    assertEquals(
        results.get(0),
        15,
        "Unsubscribe and resubscribe, progress is not retained. Full synchronization.");
    assertEquals(results.get(1), 0, "Unsubscribe and then subscribe again," + device2);
    assertEquals(results.get(2), 0, "Subscribe again after cancellation," + database2 + ".d_2");
    assertEquals(results.get(3), 3, "Number of received files: resubscribe after unsubscribe");
  }
PullConsumer
pattern: device
format: tsfile
loose-range: time
integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBPathDeviceTsfilePullConsumerIT.java