@Override
                public void run() {
                    try {
                        int getCount = msgCount;
                        ConsumeResult csmResult = new ConsumeResult();
                        ConfirmResult cfmResult = new ConfirmResult();
                        // wait partition status ready
                        do {
                            if (consumer.isPartitionsReady(5000)
                                    || consumer.isShutdown()) {
                                break;
                            }
                        } while (true);
                        // consume messages
                        do {
                            // 5.1 judge consumer is shutdown
                            if (consumer.isShutdown()) {
                                logger.info("Consumer is shutdown!");
                                break;
                            }
                            // 5.2 get messages
                            if (consumer.getMessage(csmResult)) {
                                // 5.2.1.1 process messages if success
                                List<Message> messageList = csmResult.getMessageList();
                                if (messageList != null && !messageList.isEmpty()) {
                                    msgRcvStats.addMsgCount(
                                            csmResult.getTopicName(), messageList.size());
                                }
                                // 5.2.1.2 store current offset
                                partitionOffsetMap.put(
                                        csmResult.getPartitionKey(), csmResult.getCurrOffset());
                                // 5.2.1.3 confirm messages to server
                                if (consumer.confirmConsume(
                                        csmResult.getConfirmContext(), true, cfmResult)) {
                                    // store confirmed offset
                                    partitionOffsetMap.put(
                                            cfmResult.getPartitionKey(), cfmResult.getCurrOffset());
                                }
                            } else {
                                // 5.2.2.1 print unexpected error
                                if (csmResult.getErrCode() == TErrCodeConstants.CLIENT_SHUTDOWN) {
                                    logger.info("Found that the client has shutdown, exit!");
                                }
                            }
                            // judge reached required message count
                            if (msgCount > 0) {
                                if (--getCount <= 0) {
                                    break;
                                }
                            }
                        } while (true);
                    } catch (Throwable e) {
                        logger.error("Consume messages failed!", e);
                    }
                    msgRcvStats.stopStats();
                    logger.info("Fetch runner exit!");
                }
Consume messages in Pull mode by assigning partitions by the client.

@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 totalGroupNodeCnt is the total number of clients started by the consumer group;
              The 6th  parameter fetchThreadCnt is the count of fetch thread.
inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/ClientBalanceConsumerExample.java