public Producer initTopicProducer(String topic) {
        logger.info("initTopicProducer topic = {}", topic);
        Producer producer = null;
        try {
            producer = pulsarClient.newProducer().sendTimeout(sendTimeout,
                    TimeUnit.MILLISECONDS)
                    .topic(topic)
                    .enableBatching(enableBatch)
                    .blockIfQueueFull(blockIfQueueFull)
                    .maxPendingMessages(maxPendingMessages)
                    .batchingMaxMessages(maxBatchingMessages)
                    .create();
        } catch (PulsarClientException e) {
            logger.error("create pulsar client has error e = {}", e);
        }
        return producer;
    }
If this function is called successively without calling {@see #destroyConnection()}, only the
first call has any effect.

@throws FlumeException if an RPC client connection could not be opened
inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/pulsar/PulsarClientService.java