@Override
        public void run() {
            ConsumerResult csmResult;
            ConsumerResult cfmResult;
            int getCount = consumeCount;
            // wait partition status ready
            do {
                if (pullConsumer.isPartitionsReady(5000)
                        || pullConsumer.isShutdown()) {
                    break;
                }
            } while (true);
            // consume messages
            do {
                try {
                    // 1 judge consumer is shutdown
                    if (pullConsumer.isShutdown()) {
                        logger.warn("Consumer is shutdown!");
                        break;
                    }
                    // 2 get messages from server
                    csmResult = pullConsumer.getMessage();
                    if (csmResult.isSuccess()) {
                        // 2.1 process message if getMessage() return success
                        List<Message> messageList = csmResult.getMessageList();
                        if (messageList != null && !messageList.isEmpty()) {
                            msgRcvStats.addMsgCount(csmResult.getTopicName(), messageList.size());
                        }
                        // 2.1.2 store the offset of processing message
                        // the offset returned by GetMessage() represents the initial offset of this request
                        // if consumer group is pure Pull mode, the initial offset can be saved;
                        // if not, we have to use the return value of confirmConsume()
                        partOffsetMap.put(csmResult.getPartitionKey(), csmResult.getCurrOffset());
                        // 2.1.3 confirm consume result
                        cfmResult = pullConsumer.confirmConsume(
                                csmResult.getConfirmContext(), true);
                        if (cfmResult.isSuccess()) {
                            // store confirmed offset value
                            partOffsetMap.put(csmResult.getPartitionKey(), csmResult.getCurrOffset());
                        }
                    }
                    // 3. Determine whether the consumed data reaches the goal
                    if (consumeCount > 0) {
                        if (--getCount <= 0) {
                            logger.info("Consumer has consumed {} messages!", consumeCount);
                            break;
                        }
                    }
                } catch (Throwable e) {
                    // Any exceptions in running can be ignored
                }
            } while (true);
            logger.info("The fetch thread has exited!");
        }
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