@Test
  public void do_test()
      throws InterruptedException,
          TException,
          IoTDBConnectionException,
          IOException,
          StatementExecutionException {
    consumer =
        create_pull_consumer("pull_pattern", "MiddleMatchPatternHistory_DataSet", false, null);
    // Write data before subscribing
    for (int i = 0; i < 3; i++) {
      insert_data(1706659200000L, devices.get(i)); // 2024-01-31 08:00:00+08:00
    }
    // Subscribe
    consumer.subscribe(topicName);
    assertEquals(subs.getSubscriptions().size(), 1, "show subscriptions after subscription");
    for (int i = 0; i < 3; i++) {
      insert_data(System.currentTimeMillis() - 30000L, devices.get(i));
    }
    // Consumption data
    consume_data(consumer, session_dest);
    String sql = "select count(s_0) from ";
    for (int i = 0; i < 3; i++) {
      System.out.println(
          "src " + devices.get(i) + ": " + getCount(session_src, sql + devices.get(i)));
    }
    check_count(0, sql + device, "Consumption data: s_0 " + device);
    check_count(0, "select count(s_1) from " + device, "Consumption data: s_1 " + device);
    check_count(0, "select count(s_0) from " + device2, "Consumption data: s_0 " + device2);
    check_count(0, "select count(s_1) from " + device2, "Consumption data: s_1 " + device2);
    check_count(10, "select count(s_0) from " + device3, "Consumption data: s_0 " + device3);
    check_count(10, "select count(s_1) from " + device3, "Consumption data: s_1 " + device3);
    for (int i = 0; i < 3; i++) {
      insert_data(System.currentTimeMillis(), devices.get(i));
    }
    // Unsubscribe
    consumer.unsubscribe(topicName);
    // Subscribe and then write data
    consumer.subscribe(topicName);
    assertEquals(subs.getSubscriptions().size(), 1, "show subscriptions after re-subscribing");
    for (int i = 0; i < 3; i++) {
      System.out.println(
          "src " + devices.get(i) + ": " + getCount(session_src, sql + devices.get(i)));
    }
    // Consumption data: Progress is not retained when re-subscribing after cancellation. Full
    // synchronization.
    consume_data(consumer, session_dest);
    check_count(0, "select count(s_0) from " + device, "consume data again:" + device);
    check_count(0, "select count(s_0) from " + device2, "consume data again:" + device2);
    check_count(10, "select count(s_1) from " + device3, "Consumption data:" + device3);
  }
pattern: root.*.d_*.*
format: dataset
time-range: history
integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/pattern/IoTDBMiddleMatch2PatternPullConsumerDataSetIT.java