@Override
        public void run() {
            ConsumerResult csmResult;

            // wait partition status ready
            while (true) {
                if (pullMessageConsumer.isPartitionsReady(5000) || pullMessageConsumer.isShutdown()) {
                    LOG.warn("tube partition is not ready or consumer is shutdown!");
                    break;
                }
            }
            // consume messages
            while (true) {
                if (pullMessageConsumer.isShutdown()) {
                    LOG.warn("consumer is shutdown!");
                    break;
                }

                try {
                    csmResult = pullMessageConsumer.getMessage();
                    if (csmResult.isSuccess()) {
                        List<Message> messageList = csmResult.getMessageList();
                        if (CollectionUtils.isNotEmpty(messageList)) {
                            for (Message message : messageList) {
                                if (StringUtils.equals(message.getTopic(), topic)) {
                                    String body = new String(message.getData(), StandardCharsets.UTF_8);
                                    handleMessage(body);
                                }
                            }
                        }
                        pullMessageConsumer.confirmConsume(csmResult.getConfirmContext(), true);
                    } else {
                        MetricsManager.getInstance().addReceiveFailed(1);
                        LOG.error("receive messages errorCode is {}, error meddage is {}", csmResult.getErrCode(),
                                csmResult.getErrMsg());
                    }
                } catch (TubeClientException e) {
                    MetricsManager.getInstance().addReceiveFailed(1);
                    LOG.error("tube consumer getMessage error {}", e.getMessage());
                } catch (Exception e) {
                    MetricsManager.getInstance().addReceiveFailed(1);
                    LOG.error("handle audit message error {}", e.getMessage());
                }

            }
        }
Constructor
@param insertServiceList
@param storeConfig
@param mqConfig
inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/service/consume/TubeConsume.java