private void sendData(BatchProxyMessage batchMsg) throws InterruptedException {
        if (ObjectUtils.isEmpty(batchMsg)) {
            return;
        }

        Producer producer = selectProducer();
        if (ObjectUtils.isEmpty(producer)) {
            pulsarSendQueue.put(batchMsg);
            LOGGER.error("send job[{}] data err, empty pulsar producer", jobInstanceId);
            return;
        }
        InLongMsg message = batchMsg.getInLongMsg();
        sinkMetric.pluginSendCount.addAndGet(batchMsg.getMsgCnt());
        if (asyncSend) {
            CompletableFuture<MessageId> future = producer.newMessage().eventTime(batchMsg.getDataTime())
                    .value(message.buildArray()).sendAsync();
            future.whenCompleteAsync((m, t) -> {
                if (t != null) {
                    // send error
                    sinkMetric.pluginSendFailCount.addAndGet(batchMsg.getMsgCnt());
                    LOGGER.error("send data fail to pulsar, add back to sendqueue, current queue size {}",
                            pulsarSendQueue.size(), t);
                    try {
                        pulsarSendQueue.put(batchMsg);
                    } catch (InterruptedException e) {
                        LOGGER.error("put back to queue fail send queue size {}", pulsarSendQueue.size(), t);
                    }
                } else {
                    // send succerss, update metrics
                    sendQueueSemaphore.release();
                    updateSuccessSendMetrics(batchMsg);
                }
            });
        } else {
            try {
                producer.newMessage().eventTime(batchMsg.getDataTime()).value(message.buildArray()).send();
                sendQueueSemaphore.release();
                updateSuccessSendMetrics(batchMsg);
            } catch (PulsarClientException e) {
                sinkMetric.pluginSendFailCount.addAndGet(batchMsg.getMsgCnt());
                LOGGER.error("send data fail to pulsar, add back to send queue, send queue size {}",
                        pulsarSendQueue.size(), e);
                pulsarSendQueue.put(batchMsg);
            }
        }
    }
take batchMsg from sendQueue and send to pulsar
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/PulsarSink.java