private void updateSuccessSendMetrics(BatchProxyMessage batchMsg) {
        AuditUtils.add(AuditUtils.AUDIT_ID_AGENT_SEND_SUCCESS, batchMsg.getGroupId(),
                batchMsg.getStreamId(), batchMsg.getDataTime(), batchMsg.getMsgCnt(),
                batchMsg.getTotalSize());
        sinkMetric.pluginSendSuccessCount.addAndGet(batchMsg.getMsgCnt());
    }
take batchMsg from sendQueue and send to pulsar
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/PulsarSink.java