@Override
                public void run() {
                    ConsumerResult csmResult;
                    int getCount = consumeCount;
                    // wait partition status ready
                    do {
                        if (pullConsumer.isPartitionsReady(5000)
                                || pullConsumer.isShutdown()) {
                            break;
                        }
                    } while (true);
                    // consume messages
                    do {
                        try {
                            // 4.1 judge consumer is shutdown
                            if (pullConsumer.isShutdown()) {
                                logger.warn("Consumer is shutdown!");
                                break;
                            }
                            // 4.2 get messages from server
                            csmResult = pullConsumer.getMessage();
                            if (csmResult.isSuccess()) {
                                // 4.2.1 process message if getMessage() return success
                                List<Message> messageList = csmResult.getMessageList();
                                if (messageList != null && !messageList.isEmpty()) {
                                    msgRcvStats.addMsgCount(csmResult.getTopicName(), messageList.size());
                                }
                                // 4.2.1.1 confirm consume result
                                // Notice:
                                // 1. If the processing fails, the parameter isConsumed can
                                // be set to false, but this is likely to cause
                                // an infinite loop of data consumption, so
                                // it is strongly recommended to set this parameter
                                // to true when using it, and the failed data can
                                // be processed in other ways
                                // 2. The messageList returned by getMessage() may be empty,
                                // and confirmConsume() is still required call in this case
                                pullConsumer.confirmConsume(csmResult.getConfirmContext(), true);
                            } else {
                                // 4.2.2 process failure when getMessage() return false
                                // Any failure can be ignored
                                if (!IGNORE_ERROR_SET.contains(csmResult.getErrCode())) {
                                    logger.debug(
                                            "Receive messages errorCode is {}, Error message is {}",
                                            csmResult.getErrCode(), csmResult.getErrMsg());
                                }
                            }
                            // 4.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 in Pull mode through a single-session factory instance.

@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/MessagePullConsumerExample.java