public void close() {
            if (CollectionUtils.isEmpty(producers)) {
                return;
            }
            for (Producer producer : producers) {
                try {
                    producer.close();
                } catch (Throwable e) {
                    LOGGER.error("job[{}] close pulsar producer error", jobInstanceId, e);
                }
            }
            try {
                pulsarClient.shutdown();
            } catch (PulsarClientException e) {
                LOGGER.error("job[{}] close pulsar client error", jobInstanceId, e);
            }
        }
close all pulsar producer and shutdown client
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/PulsarSink.java