private void initKafkaSender() {
        if (CollectionUtils.isEmpty(mqClusterInfos)) {
            LOGGER.error("init job[{}] kafka producer fail, empty mqCluster info", jobInstanceId);
            return;
        }
        for (MQClusterInfo clusterInfo : mqClusterInfos) {
            if (!clusterInfo.isValid()) {
                continue;
            }
            kafkaSenders.add(new KafkaSender(clusterInfo, producerNum));
        }
    }
take batchMsg from sendQueue and send to kafka
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/KafkaSink.java