private static void dataSubscription2() throws Exception {
    try (final SubscriptionTreeSession subscriptionSession =
        new SubscriptionTreeSession(HOST, PORT)) {
      subscriptionSession.open();
      final Properties config = new Properties();
      config.put(TopicConstant.START_TIME_KEY, CURRENT_TIME + 33);
      config.put(TopicConstant.END_TIME_KEY, CURRENT_TIME + 66);
      config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE);
      subscriptionSession.createTopic(TOPIC_2, config);
    }

    final List<Thread> threads = new ArrayList<>();
    for (int i = 0; i < PARALLELISM; ++i) {
      final int idx = i;
      final Thread thread =
          new Thread(
              () -> {
                int retryCount = 0;
                // Subscription: builder-style ctor
                try (final SubscriptionTreePullConsumer consumer2 =
                    new SubscriptionTreePullConsumer.Builder()
                        .consumerId("c" + idx)
                        .consumerGroupId("cg2")
                        .autoCommit(false)
                        .buildPullConsumer()) {
                  consumer2.open();
                  consumer2.subscribe(TOPIC_2);
                  while (true) {
                    final List<SubscriptionMessage> messages =
                        consumer2.poll(Collections.singleton(TOPIC_2), POLL_TIMEOUT_MS);
                    if (messages.isEmpty()) {
                      retryCount++;
                      if (retryCount >= MAX_RETRY_TIMES) {
                        break;
                      }
                    }
                    for (final SubscriptionMessage message : messages) {
                      try (final TsFileReader reader = message.getTsFileHandler().openReader()) {
                        final QueryDataSet dataSet =
                            reader.query(
                                QueryExpression.create(
                                    Arrays.asList(
                                        new Path("root.db.d2", "s2", true),
                                        new Path("root.sg.d3", "s1", true)),
                                    null));
                        while (dataSet.hasNext()) {
                          System.out.println(dataSet.next());
                        }
                      }
                    }
                    consumer2.commitSync(messages);
                  }
                  consumer2.unsubscribe(TOPIC_2);
                } catch (final IOException e) {
                  throw new RuntimeException(e);
                }
              });
      thread.start();
      threads.add(thread);
    }

    for (final Thread thread : threads) {
      thread.join();
    }
  }
multi pull consumer subscribe topic with tsfile format
example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java