private PulsarClient getOrCreatePulsarClient(String pulsarServerUrl) {
        LOG.info("start consumer pulsarServerUrl = {}", pulsarServerUrl);
        PulsarClient pulsarClient = null;
        ClientBuilder builder = PulsarClient.builder();
        try {
            if (mqConfig.isPulsarEnableAuth() && StringUtils.isNotEmpty(mqConfig.getPulsarToken())) {
                builder.authentication(AuthenticationFactory.token(mqConfig.getPulsarToken()));
            }
            pulsarClient = builder.serviceUrl(pulsarServerUrl)
                    .connectionTimeout(mqConfig.getClientOperationTimeoutSecond(), TimeUnit.SECONDS).build();
        } catch (PulsarClientException e) {
            LOG.error("getOrCreatePulsarClient has pulsar {} err {}", pulsarServerUrl, e);
        }
        return pulsarClient;
    }
Constructor
@param insertServiceList
@param storeConfig
@param mqConfig
inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/service/consume/PulsarConsume.java