public Producer getProducer() {
            if (CollectionUtils.isEmpty(producers)) {
                LOGGER.error("job[{}] empty producers", jobInstanceId);
                return null;
            }
            int index = (producerIndex.getAndIncrement() & Integer.MAX_VALUE) % producers.size();
            return producers.get(index);
        }
take batchMsg from sendQueue and send to pulsar
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/PulsarSink.java