@Override
        void run() {
            try {
                Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                    try {
                        if (clientBalanceConsumer != null)
                            clientBalanceConsumer.shutdown();
                        if (messagePullConsumer != null)
                            messagePullConsumer.shutdown();
                        if (messagePushConsumer != null)
                            messagePushConsumer.shutdown();
                    } catch (Throwable e) {
                        e.printStackTrace();
                    }
                    System.out.println(msgCount.get() + " message(s) has been consumed. Exited.");
                }));

                final ConsumerConfig consumerConfig = new ConsumerConfig(masterServers, groupName);
                switch (consumePosition) {
                    case "first":
                        consumerConfig.setConsumePosition(ConsumePosition.CONSUMER_FROM_FIRST_OFFSET);
                        break;
                    case "latest":
                        consumerConfig.setConsumePosition(ConsumePosition.CONSUMER_FROM_LATEST_OFFSET);
                        break;
                    case "max":
                        consumerConfig.setConsumePosition(ConsumePosition.CONSUMER_FROM_MAX_OFFSET_ALWAYS);
                        break;
                    default:
                        throw new ParameterException("Consume position, must in { first | latest | max }");
                }
                final MessageSessionFactory messageSessionFactory = new TubeSingleSessionFactory(consumerConfig);
                if (consumePartitionsAndOffsets != null) {
                    balanceConsumer(messageSessionFactory, consumerConfig);
                } else {
                    switch (mode) {
                        case "pull":
                            pullConsumer(messageSessionFactory, consumerConfig);
                            break;
                        case "push":
                            pushConsumer(messageSessionFactory, consumerConfig);
                            break;
                        case "balance":
                            balanceConsumer(messageSessionFactory, consumerConfig);
                            break;
                        default:
                            throw new ParameterException("Consume mode, must in { pull | push | balance }");
                    }
                }

            } catch (Exception e) {
                System.out.println(e.getMessage());
            } catch (Throwable e) {
                e.printStackTrace();
            }
        }
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