在没有TrustStore / KeyStore的情况下使用Java连接到SASL Kafka broker

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaSASLExample {
    public static void main(String[] args) {
        // Kafka broker地址
        String bootstrapServers = "kafka-broker1:9092,kafka-broker2:9092";

        // SASL认证配置
        String saslMechanism = "PLAIN";
        String securityProtocol = "SASL_PLAINTEXT";
        String saslJaasConfig = "org.apache.kafka.common.security.plain.PlainLoginModule required " +
                "username=\"your-username\" password=\"your-password\";";

        // 创建Kafka Producer配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put("security.protocol", securityProtocol);
        properties.put("sasl.mechanism", saslMechanism);
        properties.put("sasl.jaas.config", saslJaasConfig);

        // 创建Kafka Producer
        Producer<String, String> producer = new KafkaProducer<>(properties);

        // 发送消息
        producer.send(new ProducerRecord<>("your-topic", "your-key", "your-value"));

        // 关闭Kafka Producer
        producer.close();
    }
}