private void initPulsarSender() {
        if (CollectionUtils.isEmpty(mqClusterInfos)) {
            LOGGER.error("init job[{}] pulsar client fail, empty mqCluster info", jobInstanceId);
            return;
        }
        for (MQClusterInfo clusterInfo : mqClusterInfos) {
            if (clusterInfo.isValid()) {
                try {
                    PulsarClient client = PulsarClient.builder().serviceUrl(clusterInfo.getUrl())
                            .ioThreads(clientIoThreads)
                            .connectionsPerBroker(connectionsPreBroker).build();
                    pulsarSenders.add(new PulsarTopicSender(client, producerNum));
                    LOGGER.info("job[{}] init pulsar client url={}", jobInstanceId, clusterInfo.getUrl());
                } catch (PulsarClientException e) {
                    LOGGER.error("init job[{}] pulsar client fail", jobInstanceId, e);
                }
            }
        }
    }
take batchMsg from sendQueue and send to pulsar
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/PulsarSink.java