private void pullConsumer(MessageSessionFactory messageSessionFactory, ConsumerConfig consumerConfig)
                throws TubeClientException {
            messagePullConsumer = messageSessionFactory.createPullConsumer(consumerConfig);
            messagePullConsumer.subscribe(topicName, null);
            messagePullConsumer.completeSubscribe();
            while (!messagePullConsumer.isPartitionsReady(1000)) {
                ThreadUtils.sleep(1000);
            }
            System.out.println("Ready to consume messages......");
            while (true) {
                ConsumerResult result = messagePullConsumer.getMessage();
                if (result.isSuccess()) {
                    List<Message> messageList = result.getMessageList();
                    for (Message message : messageList) {
                        System.out.println(new String(message.getData()));
                        msgCount.getAndIncrement();
                    }
                    messagePullConsumer.confirmConsume(result.getConfirmContext(), true);
                }
            }
        }
Create a pullConsumer 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