private Producer selectProducer() {
        if (CollectionUtils.isEmpty(pulsarSenders)) {
            LOGGER.error("send job[{}] data err, empty pulsar sender", jobInstanceId);
            return null;
        }
        PulsarTopicSender sender = pulsarSenders.get(
                (CLIENT_INDEX.getAndIncrement() & Integer.MAX_VALUE) % pulsarSenders.size());
        return sender.getProducer();
    }
take batchMsg from sendQueue and send to pulsar
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/PulsarSink.java