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

        KafkaProducer<String, byte[]> producer = selectProducer();
        if (ObjectUtils.isEmpty(producer)) {
            kafkaSendQueue.put(batchMsg);
            LOGGER.error("send job[{}] data err, empty kafka producer", jobInstanceId);
            return;
        }

        ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic, batchMsg.getInLongMsg().buildArray());
        sinkMetric.pluginSendCount.addAndGet(batchMsg.getMsgCnt());
        if (asyncSend) {
            producer.send(record, new AsyncSinkCallback(System.currentTimeMillis(), batchMsg));
        } else {
            try {
                Future<RecordMetadata> future = producer.send(record);
                future.get(DEFAULT_KAFKA_SINK_SYNC_SEND_TIMEOUT_MS, TimeUnit.MILLISECONDS);
                updateSuccessSendMetrics(batchMsg);
            } catch (Exception e) {
                sinkMetric.pluginSendFailCount.addAndGet(batchMsg.getMsgCnt());
                LOGGER.error("send job[{}] data fail to kafka, add back to send queue, send queue size {}",
                        jobInstanceId,
                        kafkaSendQueue.size(), e);
                kafkaSendQueue.put(batchMsg);
            }
        }
    }
take batchMsg from sendQueue and send to kafka
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/KafkaSink.java