private static PullMessageConsumer createPullConsumer(
            String masterHostAndPorts, String groupName) throws Exception {
        // 1. initial configure and build session factory object
        ConsumerConfig consumerConfig =
                new ConsumerConfig(masterHostAndPorts, groupName);
        // 1.2 set consume from latest position if the consumer group is first consume
        consumerConfig.setConsumePosition(ConsumePosition.CONSUMER_FROM_LATEST_OFFSET);
        // 2. build session factory object
        // find and initialize TubeMultiSessionFactory object according to the Master cluster information
        MasterInfo masterInfo = consumerConfig.getMasterInfo();

        MessageSessionFactory sessionFactory;
        synchronized (multSessFtyMap) {
            sessionFactory =
                    multSessFtyMap.get(masterInfo.getMasterClusterStr());
            if (sessionFactory == null) {
                sessionFactory = new TubeMultiSessionFactory(consumerConfig);
                multSessFtyMap.put(masterInfo.getMasterClusterStr(), sessionFactory);
            }
        }

        // 3. Create and get the PullMessageConsumer object
        return sessionFactory.createPullConsumer(consumerConfig);
    }
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