@Test
  public void do_test()
      throws InterruptedException,
          TException,
          IoTDBConnectionException,
          IOException,
          StatementExecutionException {
    // Write data before subscribing
    session_src.executeNonQueryStatement(
        "insert into "
            + device
            + "(time,s_0,s_1)values(1710288000000,313,'2024-03-13 08:00:00+08:00');"); // 2024-03-13
    // 08:00:00+08:00
    session_src.executeNonQueryStatement(
        "insert into "
            + device
            + "(time,s_0,s_1)values(1703980800000,133.45,'2023-12-31 08:00:00+08:00');"); // 2023-12-31 08:00:00+08:00
    insert_data(1706659200000L, device); // 2024-01-31 08:00:00+08:00
    insert_data(1706659200000L, device2); // 2024-01-31 08:00:00+08:00
    AtomicInteger rowCount1 = new AtomicInteger(0);
    AtomicInteger rowCount2 = new AtomicInteger(0);
    // Subscribe
    consumer =
        new SubscriptionTreePushConsumer.Builder()
            .host(SRC_HOST)
            .port(SRC_PORT)
            .consumerId("db_mix_consumer_2_topic")
            .consumerGroupId("OneConsumerMultiTopicsMix")
            .ackStrategy(AckStrategy.AFTER_CONSUME)
            .fileSaveDir("target/push-subscription")
            .consumeListener(
                message -> {
                  final short messageType = message.getMessageType();
                  if (SubscriptionMessageType.isValidatedMessageType(messageType)) {
                    switch (SubscriptionMessageType.valueOf(messageType)) {
                      case SESSION_DATA_SETS_HANDLER:
                        for (final SubscriptionSessionDataSet dataSet :
                            message.getSessionDataSetsHandler()) {
                          try {
                            session_dest.insertTablet(dataSet.getTablet());
                          } catch (StatementExecutionException e) {
                            throw new RuntimeException(e);
                          } catch (IoTDBConnectionException e) {
                            throw new RuntimeException(e);
                          }
                        }
                        break;
                      case TS_FILE_HANDLER:
                        try {
                          TsFileReader reader = message.getTsFileHandler().openReader();
                          QueryDataSet dataset =
                              reader.query(
                                  QueryExpression.create(
                                      Collections.singletonList(new Path(device, "s_0", true)),
                                      null));
                          while (dataset.hasNext()) {
                            rowCount1.addAndGet(1);
                            RowRecord next = dataset.next();
                            System.out.println(
                                device + ":" + next.getTimestamp() + "," + next.getFields());
                          }
                          dataset =
                              reader.query(
                                  QueryExpression.create(
                                      Collections.singletonList(new Path(device2, "s_0", true)),
                                      null));
                          while (dataset.hasNext()) {
                            rowCount2.addAndGet(1);
                            RowRecord next = dataset.next();
                            System.out.println(
                                device2 + ":" + next.getTimestamp() + "," + next.getFields());
                          }
                        } catch (IOException e) {
                          throw new RuntimeException(e);
                        }
                        break;
                    }
                  }
                  return ConsumeResult.SUCCESS;
                })
            .buildPushConsumer();
    consumer.open();
    consumer.subscribe(topicName, topicName2);

    System.out.println("###### Subscription Query:");
    subs.getSubscriptions().forEach((System.out::println));
    assertEquals(subs.getSubscriptions().size(), 2, "subscribe and show subscriptions");
    // Subscribe and then write data
    Thread thread =
        new Thread(
            () -> {
              try {
                insert_data(System.currentTimeMillis(), device);
                insert_data(System.currentTimeMillis(), device2);
              } catch (Exception e) {
                throw new RuntimeException(e);
              }
            });
    thread.start();
    thread.join();

    AWAIT.untilAsserted(
        () -> {
          assertEquals(rowCount1.get(), 0, "pattern1");
          check_count(12, "select count(s_0) from " + device, "dataset pattern1");
          assertEquals(rowCount2.get(), 10, "pattern2");
        });
  }
1 consumer subscribes to 2 topics: historical data
The timing of flush is very critical. If the data inside the filter and the data outside the filter are within one tsfile, they will all be extracted.
integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsMixIT.java