protected Consumer<byte[]> createConsumer(PulsarClient pulsarClient, String topic) {
        Consumer<byte[]> consumer = null;
        if (pulsarClient != null && StringUtils.isNotEmpty(topic)) {
            LOG.info("createConsumer has topic {}, subName {}", topic,
                    mqConfig.getPulsarConsumerSubName());
            try {
                consumer = pulsarClient.newConsumer()
                        .subscriptionName(mqConfig.getPulsarConsumerSubName())
                        .subscriptionType(SubscriptionType.Shared)
                        .topic(topic)
                        .receiverQueueSize(mqConfig.getConsumerReceiveQueueSize())
                        .enableRetry(mqConfig.isPulsarConsumerEnableRetry())
                        .messageListener(new MessageListener<byte[]>() {

                            public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
                                String body = null;
                                try {
                                    body = new String(msg.getData(), StandardCharsets.UTF_8);
                                    handleMessage(body, consumer, msg.getMessageId());
                                } catch (Exception exception) {
                                    MetricsManager.getInstance().addReceiveFailed(1);
                                    PulsarUtils.acknowledge(consumer, msg.getMessageId());
                                    LOG.error("Invalid audit data: {}", body, exception);
                                }
                            }
                        })
                        .subscribe();
            } catch (PulsarClientException e) {
                LOG.error("createConsumer has topic {}, subName {}, err {}", topic,
                        mqConfig.getPulsarConsumerSubName(), e);
            }
        }
        return consumer;
    }
Constructor
@param insertServiceList
@param storeConfig
@param mqConfig
inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/service/consume/PulsarConsume.java