@Test
  public void do_test()
      throws InterruptedException,
          TException,
          IoTDBConnectionException,
          IOException,
          StatementExecutionException {
    consumer = create_pull_consumer("tsfile_group_share_process", "c1", false, null);
    consumer2 = create_pull_consumer("tsfile_group_share_process", "c2", false, 10L);
    // Subscribe
    assertEquals(
        subs.getSubscriptions(topicName).size(), 0, "Show subscriptions before subscribing");
    consumer.subscribe(topicName);
    consumer2.subscribe(topicName);
    subs.getSubscriptions(topicName).forEach((System.out::println));
    assertEquals(
        subs.getSubscriptions(topicName).size(), 1, "show subscriptions after subscription");
    // insert 1000 records
    Thread thread =
        new Thread(
            () -> {
              long timestamp = 1706659200000L; // 2024-01-31 08:00:00+08:00
              for (int i = 0; i < 20; i++) {
                try {
                  insert_data(timestamp);
                } catch (IoTDBConnectionException e) {
                  throw new RuntimeException(e);
                } catch (StatementExecutionException e) {
                  throw new RuntimeException(e);
                }
                timestamp += 20000;
              }
            });
    AtomicInteger rowCount1 = new AtomicInteger(0);
    AtomicInteger rowCount2 = new AtomicInteger(0);
    Thread thread1 =
        new Thread(
            () -> {
              try {
                rowCount1.addAndGet(consume_tsfile(consumer, device));
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            });
    Thread thread2 =
        new Thread(
            () -> {
              try {
                rowCount2.addAndGet(consume_tsfile(consumer2, device));
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            });

    thread1.start();
    thread2.start();
    thread.start();
    thread1.join();
    thread2.join();
    thread.join();

    System.out.println("src=" + getCount(session_src, "select count(s_0) from " + device));
    System.out.println("rowCount1=" + rowCount1.get());
    System.out.println("rowCount2=" + rowCount2.get());
    AWAIT.untilAsserted(
        () ->
            assertGte(
                rowCount1.get() + rowCount2.get(),
                getCount(session_src, "select count(s_0) from " + device),
                "consumer share process rowCount1="
                    + rowCount1.get()
                    + " rowCount2="
                    + rowCount2.get()
                    + " src="
                    + getCount(session_src, "select count(s_0) from " + device)));
  }
format: tsfile
pattern:device
Same group pull consumer share progress
integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/multi/IoTDBConsumer2With1TopicShareProcessTsfileIT.java