public static void main(String[] args) throws Throwable {
        // 1. get and initial parameters
        final String masterServers = args[0];
        final String subTopicAndFiterItems = args[1];
        final String groupName = args[2];
        final int consumeCount = Integer.parseInt(args[3]);
        int fetchThreadCnt = 3;
        if (args.length > 4) {
            fetchThreadCnt = MixedUtils.mid(Integer.parseInt(args[4]),
                    1, Runtime.getRuntime().availableProcessors());
        }
        final Map<String, TreeSet<String>> topicAndFiltersMap =
                MixedUtils.parseTopicParam(subTopicAndFiterItems);

        // 2. initial and statistic thread
        Thread statisticThread =
                new Thread(msgRcvStats, "Receive Statistic Thread");
        statisticThread.start();

        // 3. Start the consumer group for the first consumption
        // 3.1. build consumer object
        logger.info("The first consumption begin!");
        PullMessageConsumer consumer1 =
                createPullConsumer(masterServers, groupName);
        // 3.2. Set the Topic and the filter item set corresponding to the consumption
        // if you not need filter consumption,
        // set the parameter filterConds is null or empty set
        for (Map.Entry<String, TreeSet<String>> entry : topicAndFiltersMap.entrySet()) {
            consumer1.subscribe(entry.getKey(), entry.getValue());
        }
        // 3.3. start consumption with
        String sessionKeyFst = "test_consume_first";
        int sourceCountFst = 1;
        boolean isSelectBig = false;
        // The map of partitionKey and last success offset
        // You can persist the information and use it when restarting or
        // re-rolling to keep the current consumption to start from
        // the offset required in the last round
        ConcurrentHashMap<String, Long> partOffsetMapFst =
                new ConcurrentHashMap<>();
        consumer1.completeSubscribe(sessionKeyFst,
                sourceCountFst, isSelectBig, partOffsetMapFst);

        // 3.4. initial and start fetch threads
        Thread[] fetchRunners1 = new Thread[fetchThreadCnt];
        for (int i = 0; i < fetchRunners1.length; i++) {
            fetchRunners1[i] = new Thread(new FetchRequestRunner(
                    consumer1, partOffsetMapFst, consumeCount), "_fetch_runner_" + i);
        }
        for (Thread thread : fetchRunners1) {
            thread.start();
        }

        // 3.5. wait consume data
        ThreadUtils.sleep(2 * 60 * 1000);
        logger.info("The first consumption has finished!");

        // 3.6. shutdown consumer
        consumer1.shutdown();
        for (Thread thread : fetchRunners1) {
            thread.join();
        }

        // 4. Start the consumer group for the second consumption
        // 4.1 set the boostrap Offset, here we set consumption from 0
        logger.info("The second consumption begin!");
        String sessionKeySec = "test_consume_Second";
        int sourceCountSec = 1;
        ConcurrentHashMap<String, Long> partOffsetMapSec =
                new ConcurrentHashMap<>();
        for (String partKey : partOffsetMapFst.keySet()) {
            partOffsetMapSec.put(partKey, 0L);
        }
        // 4.2. build consumer object
        PullMessageConsumer consumer2 =
                createPullConsumer(masterServers, groupName);
        // 4.3. Set the Topic and the filter item set corresponding to the consumption
        for (Map.Entry<String, TreeSet<String>> entry : topicAndFiltersMap.entrySet()) {
            consumer2.subscribe(entry.getKey(), entry.getValue());
        }
        consumer2.completeSubscribe(sessionKeySec,
                sourceCountSec, isSelectBig, partOffsetMapSec);

        // 4.4. initial and start fetch threads
        Thread[] fetchRunners2 = new Thread[fetchThreadCnt];
        for (int i = 0; i < fetchRunners2.length; i++) {
            fetchRunners2[i] = new Thread(new FetchRequestRunner(
                    consumer2, partOffsetMapSec, consumeCount), "_fetch_runner_" + i);
        }
        for (Thread thread : fetchRunners2) {
            thread.start();
        }
    }
Consume messages with setting bootstrap offset in Pull mode through multi-session factory instances.

@param args   Startup parameter array, including the following parts:
              The 1st parameter masterServers is the master address(es) to connect to,
                      format is master1_ip:port[,master2_ip:port];
              The 2nd parameter subTopicAndFiterItems the topic(s) (and filter condition set) to consume on,
                      format is topic_1[[:filterCond_1.1[;filterCond_1.2]][,topic_2]];
              The 3rd parameter groupName is the name of consumer group;
              The 4th parameter consumeCount is the amount of messages that need to be consumed;
              The 5th  parameter fetchThreadCnt is the count of fetch thread.
inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MessagePullSetConsumerExample.java