@Test
  public void do_test()
      throws InterruptedException,
          TException,
          IoTDBConnectionException,
          IOException,
          StatementExecutionException {
    String sql = "select count(s_0) from " + device;
    consumer = create_pull_consumer("device_pattern_dataset_pull", "loose_range_path", false, null);
    // 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
    // Subscribe
    consumer.subscribe(topicName);
    assertEquals(subs.getSubscriptions().size(), 1, "show subscriptions after subscription");
    insert_data(System.currentTimeMillis() - 30000L, device);
    insert_data(System.currentTimeMillis() - 30000L, device2);
    // Consumption data
    consume_data(consumer, session_dest);
    System.out.println("src: " + getCount(session_src, sql));
    check_count(10, sql, "Consumption data: s_0 " + device);
    check_count(10, "select count(s_1) from " + device, "Consumption data: s_1");
    check_count(0, "select count(s_0) from " + device2, "Consumption data:d_1");
    check_count(0, "select count(s_0) from " + database2 + ".d_2", "Consumption data:d_2");
    insert_data(System.currentTimeMillis(), device);
    insert_data(System.currentTimeMillis(), device2);
    // Unsubscribe
    consumer.unsubscribe(topicName);
    // 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
    System.out.println("src: " + getCount(session_src, sql));
    // Consumption data: Progress is not retained when re-subscribing after cancellation. Full
    // synchronization.
    consume_data(consumer, session_dest);
    check_count(15, "select count(s_0) from " + device, "Consume data again:" + pattern);
    check_count(15, "select count(s_1) from " + device, "Consumption data: s_1");
    check_count(0, "select count(s_0) from " + device2, "Consumption data:d_1");
    check_count(0, "select count(s_0) from " + database2 + ".d_2", "Consumption data:d_2");
  }
pull consumer
format: dataset
loose-range: path
integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBPathDeviceDataSetPullConsumerIT.java