private void setKafkaProps(MQClusterInfo clusterInfo) {
            kafkaProps.clear();
            Map<String, String> params = clusterInfo.getParams();

            // set bootstrap server
            String bootStrapServers = params.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG);
            if (bootStrapServers == null) {
                throw new IllegalArgumentException("kafka param bootstrap.servers is null");
            }
            kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);

            // set ack
            String acks = params.get(ProducerConfig.ACKS_CONFIG);
            if (StringUtils.isNotEmpty(acks)) {
                kafkaProps.put(ProducerConfig.ACKS_CONFIG, acks);
            } else {
                kafkaProps.put(ProducerConfig.ACKS_CONFIG, DEFAULT_KAFKA_SINK_SEND_ACKS);
            }

            // set compression
            String compressionType = params.get(ProducerConfig.COMPRESSION_TYPE_CONFIG);
            if (StringUtils.isNotEmpty(compressionType)) {
                kafkaProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionType);
            } else {
                kafkaProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, DEFAULT_KAFKA_SINK_SEND_COMPRESSION_TYPE);
            }

            // set serializer
            String keySerializer = params.get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG);
            if (StringUtils.isNotEmpty(keySerializer)) {
                kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
            } else {
                kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, DEFAULT_KAFKA_SINK_SEND_KEY_SERIALIZER);
            }

            String valueSerializer = params.get(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG);
            if (StringUtils.isNotEmpty(keySerializer)) {
                kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
            } else {
                kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, DEFAULT_KAFKA_SINK_SEND_VALUE_SERIALIZER);
            }

            // set linger
            String lingerMs = params.get(ProducerConfig.LINGER_MS_CONFIG);
            if (StringUtils.isNotEmpty(lingerMs)) {
                kafkaProps.put(ProducerConfig.LINGER_MS_CONFIG, Integer.parseInt(lingerMs));
            }

            // set batch size
            String batchSize = params.get(ProducerConfig.BATCH_SIZE_CONFIG);
            if (StringUtils.isNotEmpty(batchSize)) {
                kafkaProps.put(ProducerConfig.BATCH_SIZE_CONFIG, Integer.parseInt(batchSize));
            }

            // set buffer memory
            String bufferMemory = params.get(ProducerConfig.BUFFER_MEMORY_CONFIG);
            if (StringUtils.isNotEmpty(batchSize)) {
                kafkaProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, Integer.parseInt(bufferMemory));
            }

            // set authentication
            String securityProtocol = params.get(SECURITY_PROTOCOL_CONFIG);
            if (StringUtils.isNotEmpty(securityProtocol)) {
                kafkaProps.put(SECURITY_PROTOCOL_CONFIG, securityProtocol);
            }

            String saslMechanism = params.get(SASL_MECHANISM);
            if (StringUtils.isNotEmpty(saslMechanism)) {
                kafkaProps.put(SASL_MECHANISM, saslMechanism);
            }

            String saslJaasConfig = params.get(SASL_JAAS_CONFIG);
            if (StringUtils.isNotEmpty(saslJaasConfig)) {
                kafkaProps.put(SASL_JAAS_CONFIG, saslJaasConfig);
            }
        }
take batchMsg from sendQueue and send to kafka
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/KafkaSink.java