Reactor Kafka消费需认证Kafka Topic在GCP环境认证失败求助

从错误信息和你的描述来看，核心矛盾是本地环境正常、GCP环境认证阶段连接终止，但命令行工具能正常消费，说明问题大概率出在代码的安全配置与GCP Kafka集群要求不匹配上，以下是具体排查点和解决方案：

1. 核心问题：安全协议与端口不匹配 #

错误中显示连接的是9093端口，这个端口在大多数Kafka部署中对应SASL_SSL协议（加密+认证），但你的代码中设置的是SASL_PLAINTEXT（明文+认证）。本地环境可能使用的是明文端口（如9092），所以正常；GCP集群的9093端口要求加密传输，导致认证阶段连接中断。

解决方法 #

修改安全协议配置为SASL_SSL：

val options = receiverOptions
    // 保留其他配置
    .consumerProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL") // 替换原SASL_PLAINTEXT

2. 补充SSL信任配置（按需） #

GCP托管的Kafka集群通常使用公网可信CA证书，你可以直接复用JVM默认的信任库，避免手动配置证书：

options
    // 追加SSL信任库配置
    .consumerProperty(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, 
                      System.getProperty("java.home") + "/lib/security/cacerts")
    .consumerProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "changeit") // JVM默认信任库密码

3. 排查默认配置冲突 #

检查defaultKafkaBrokerConfig.getAsProperties()中是否包含与安全相关的属性（如security.protocol、sasl.*），这些默认配置可能覆盖你代码中显式设置的值。可以通过打印配置来确认：

println("Final consumer properties: ${options.consumerProperties()}")

4. 验证JAAS配置格式（可选） #

虽然命令行能正常连接，但可以尝试调整JAAS配置的引号格式，避免潜在的转义问题：

val kafkaJaasConfig = "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"$kafkaUsername\" password=\"$kafkaPassword\";"

5. 其他排查点 #

• 确认环境变量正确性：在GCP环境中打印kafkaUsername和kafkaPassword的脱敏值，确保注入的凭证与命令行使用的一致。
• 客户端版本兼容：确保你的Reactor Kafka依赖的Kafka客户端版本与GCP Kafka集群版本兼容（SCRAM-SHA-512要求Broker版本≥0.10.2.0）。

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