private Producer<byte[]> createProducer() {
            try {
                return pulsarClient.newProducer().topic(topic)
                        .sendTimeout(sendTimeoutSecond, TimeUnit.SECONDS)
                        .topic(topic)
                        .enableBatching(enableBatch)
                        .blockIfQueueFull(blockIfQueueFull)
                        .maxPendingMessages(maxPendingMessages)
                        .maxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions)
                        .compressionType(compressionType)
                        .batchingMaxMessages(maxBatchingMessages)
                        .batchingMaxBytes(maxBatchingBytes)
                        .batchingMaxPublishDelay(maxBatchingPublishDelayMillis, TimeUnit.MILLISECONDS)
                        .create();
            } catch (Throwable e) {
                LOGGER.error("job[{}] create producer[topic:{}] error", jobInstanceId, topic, e);
                return null;
            }
        }
close all pulsar producer and shutdown client
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/PulsarSink.java