Azure上Confluent Cloud Java消费者认证失败问题求助

针对你遇到的SaslAuthenticationException: Authentication failed错误，结合你的代码，从以下几个方面排查修复：

1. 修正JAAS配置的语法与拼写错误 #

你的代码中SASL_JAAS_CONFIG配置存在两处问题：

• 字符串未闭合：配置行末尾缺少闭合双引号
• 拼写错误：secreteValue应为secretValue

正确的配置格式如下：

props.put(SaslConfigs.SASL_JAAS_CONFIG, 
    "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + "你的API Key值" + "\" password=\"" + "你的API Secret值" + "\";");

同时确保API Key和Secret完全复制自Confluent Cloud控制台，无多余空格或特殊字符。

2. 确认依赖版本兼容性 #

使用与Confluent Cloud兼容的Kafka客户端版本，建议选用Confluent Platform对应的kafka-clients版本（如7.x系列），避免版本不兼容导致的SASL协议适配问题。

Maven依赖示例：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>7.5.0</version>
</dependency>

3. 验证API Key的权限 #

登录Confluent Cloud控制台，检查你的API Key是否具备：

• 与目标Kafka集群的关联权限
• 目标Topic的读取权限
• 所属ACL允许执行Consumer操作

4. 补充SSL信任库配置（可选） #

部分环境下Java默认信任库无法识别Confluent Cloud的SSL证书，可显式配置系统信任库：

props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, 
    System.getProperty("java.home") + "/lib/security/cacerts");
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "changeit");

修正后的完整示例代码 #

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.CommonClientConfigs;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class HelloConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "你的应用ID");
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "XXXXXXXXXX.azure.confluent.cloud:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "someid");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        
        // 安全配置
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        props.put(SaslConfigs.SASL_JAAS_CONFIG, 
            "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"你的API Key\" password=\"你的API Secret\";");
        
        // 可选SSL信任库配置
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, 
            System.getProperty("java.home") + "/lib/security/cacerts");
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "changeit");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
        kafkaConsumer.subscribe(Arrays.asList("test-topic"));

        while(true){
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records){
                System.out.println("Key: " + record.key() + ", Value: " + record.value());
                System.out.println("Partition: " + record.partition() + ", Offset:" + record.offset());
            }
        }
    }
}

内容的提问来源于stack exchange，提问作者PAA