以编程方式验证合流Kafka访问权限

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class KafkaAccessValidator {
    public static void main(String[] args) {
        // Kafka集群连接参数配置
        String bootstrapServers = "kafka.example.com:9092";
        String securityProtocol = "SASL_SSL";
        String saslMechanism = "PLAIN";
        String username = "your-username";
        String password = "your-password";

        // 创建Kafka访问凭证
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("security.protocol", securityProtocol);
        props.put("sasl.mechanism", saslMechanism);
        props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + username + "\" password=\"" + password + "\";");
        
        // 建立与Kafka集群的连接
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        // 发送测试消息
        ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "test-key", "test-value");
        try {
            producer.send(record).get();
            System.out.println("Access permission validated successfully.");
        } catch (Exception e) {
            System.out.println("Access permission validation failed: " + e.getMessage());
        } finally {
            producer.close();
        }
    }
}