private Runnable sendDataThread() {
        return () -> {
            LOGGER.info("start pulsar sink send data thread, job[{}], groupId[{}]", jobInstanceId, inlongGroupId);
            while (!shutdown) {
                try {
                    BatchProxyMessage data = pulsarSendQueue.poll(1, TimeUnit.MILLISECONDS);
                    if (ObjectUtils.isEmpty(data)) {
                        continue;
                    }
                    sendData(data);
                } catch (Throwable t) {
                    LOGGER.error("Send data error", t);
                }
            }
        };
    }
take batchMsg from sendQueue and send to pulsar
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/PulsarSink.java