private void balanceConsumer(MessageSessionFactory messageSessionFactory, ConsumerConfig consumerConfig)
                throws TubeClientException {
            clientBalanceConsumer = messageSessionFactory.createBalanceConsumer(consumerConfig);
            ProcessResult procResult = new ProcessResult();
            QueryMetaResult qryResult = new QueryMetaResult();
            final Map<String, TreeSet<String>> topicAndFiltersMap =
                    MixedUtils.parseTopicParam(topicName);
            if (!clientBalanceConsumer.start(topicAndFiltersMap, -1, 0, procResult)) {
                System.out.println("Initial balance consumer failure, errcode is " + procResult.getErrCode()
                        + " errMsg is " + procResult.getErrMsg());
                return;
            }
            clientBalanceConsumer.getPartitionMetaInfo(qryResult);
            Map<String, Boolean> partMetaInfoMap = qryResult.getPartStatusMap();
            if (partMetaInfoMap != null && !partMetaInfoMap.isEmpty()) {
                Set<String> configuredTopicPartitions = partMetaInfoMap.keySet();
                // parse the consumePartitionsAndOffsets parameters
                Map<Long, Long> assignedPartitionsAndOffsets = new HashMap<>();
                for (String str : consumePartitionsAndOffsets.split(",")) {
                    String[] splits = str.split(":");
                    assignedPartitionsAndOffsets.put(Long.parseLong(splits[0]), Long.parseLong(splits[1]));
                }
                Set<Long> assignedPartitionIds = assignedPartitionsAndOffsets.keySet();
                Set<String> assignedPartitions = new HashSet<>();
                for (String partKey : configuredTopicPartitions) {
                    long parId = Long.parseLong(partKey.split(":")[2]);
                    if (partMetaInfoMap.get(partKey) && assignedPartitionIds.contains(parId)) {
                        assignedPartitions.add(partKey);
                        Long boostrapOffset = assignedPartitionsAndOffsets.get(parId);
                        // connect to the partitions based on consumePartitionsAndOffsets parameters
                        if (!clientBalanceConsumer.connect2Partition(partKey,
                                boostrapOffset == null ? -1L : boostrapOffset, procResult))
                            System.out.println("connect2Partition failed.");
                    }
                }

                ConsumeResult csmResult = new ConsumeResult();
                ConfirmResult cfmResult = new ConfirmResult();
                while (!clientBalanceConsumer.isPartitionsReady(1000)) {
                    ThreadUtils.sleep(1000);
                }
                System.out.println("Ready to consume messages......");
                while (true) {
                    // get messages
                    if (clientBalanceConsumer.getMessage(csmResult)) {
                        List<Message> messageList = csmResult.getMessageList();
                        for (Message message : messageList) {
                            System.out.println(new String(message.getData()));
                            msgCount.getAndIncrement();
                        }
                        // confirm messages to server
                        clientBalanceConsumer.confirmConsume(csmResult.getConfirmContext(), true, cfmResult);
                    }
                }

            } else {
                System.out.println("No partitions of the topic are available now.");
            }

        }
Create a clientBalanceConsumer and consume messages

@param messageSessionFactory
@param consumerConfig
@throws TubeClientException
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/MessageCommand.java