Confluent Kafka OAuth2 Client Credentials配置异常及正确配置咨询

一、Kafka服务端核心配置（确保认证生效的关键） #

1. 启用SASL-OAuth2认证 #

在server.properties中配置以下核心参数，必须使用SASL_SSL或SASL_PLAINTEXT协议（生产环境优先SASL_SSL）：

# 映射协议与安全类型
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
# 指定启用OAuth2的监听地址，替换为你的broker地址
listeners=SASL_SSL://0.0.0.0:9093
# 集群内部通信使用的协议
inter.broker.listener.name=SASL_SSL
# 启用OAUTHBEARER机制
sasl.enabled.mechanisms=OAUTHBEARER
sasl.mechanism.inter.broker.protocol=OAUTHBEARER

2. 对接Keycloak令牌验证 #

添加Keycloak相关的令牌校验配置，所有参数必须与Keycloak实际配置匹配：

# 指定OAuth2令牌验证的回调处理类
listener.name.sasl_ssl.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.security.oauthbearer.OAuthBearerValidatorCallbackHandler
# Keycloak的JWKS端点，用于验证令牌签名
listener.name.sasl_ssl.oauthbearer.jwks.endpoint.url=https://<你的Keycloak域名>/auth/realms/<你的Realm>/protocol/openid-connect/certs
# Keycloak的令牌 introspect 端点，用于校验令牌有效性
listener.name.sasl_ssl.oauthbearer.token.endpoint.url=https://<你的Keycloak域名>/auth/realms/<你的Realm>/protocol/openid-connect/token/introspect
# Keycloak中有权限调用introspect接口的客户端ID（需开启Service Accounts Enabled）
listener.name.sasl_ssl.oauthbearer.client.id=<合法的Keycloak客户端ID>
# 对应客户端的Secret
listener.name.sasl_ssl.oauthbearer.client.secret=<合法的Keycloak客户端Secret>
# 指定令牌中存储scope的Claim名称
listener.name.sasl_ssl.oauthbearer.scope.claim.name=scope
# Keycloak的合法Issuer URI
listener.name.sasl_ssl.oauthbearer.valid.issuer.uri=https://<你的Keycloak域名>/auth/realms/<你的Realm>

3. 可选：启用ACL授权（增强安全性） #

如果需要基于客户端身份控制资源访问，添加以下配置：

# 使用Confluent的SimpleACL授权器
authorizer.class.name=io.confluent.kafka.security.auth.SimpleAclAuthorizer
# 关闭默认允许所有请求，必须显式配置ACL规则
allow.everyone.if.no.acl.found=false

示例：给指定客户端授权写入test-topic的权限：

kafka-acls.sh --authorizer-properties zookeeper.connect=<ZK地址>:2181 --add --allow-principal User:<客户端ID> --operation Write --topic test-topic

二、客户端（生产者/消费者）配置 #

1. 生产者配置（producer.properties） #

bootstrap.servers=<Kafka Broker地址>:9093
security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
# JAAS配置，替换为你的Keycloak客户端信息
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  clientId="<合法的Keycloak客户端ID>" \
  clientSecret="<合法的Keycloak客户端Secret>" \
  scope="<你的Scope>" \
  tokenEndpointUrl="https://<你的Keycloak域名>/auth/realms/<你的Realm>/protocol/openid-connect/token";
# 指定OAuth2登录回调处理类
sasl.login.callback.handler.class=io.confluent.kafka.security.oauthbearer.OAuthBearerLoginCallbackHandler

2. 消费者配置（consumer.properties） #

bootstrap.servers=<Kafka Broker地址>:9093
group.id=test-consumer-group
security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
# JAAS配置，替换为你的Keycloak客户端信息
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  clientId="<合法的Keycloak客户端ID>" \
  clientSecret="<合法的Keycloak客户端Secret>" \
  scope="<你的Scope>" \
  tokenEndpointUrl="https://<你的Keycloak域名>/auth/realms/<你的Realm>/protocol/openid-connect/token";
# 指定OAuth2登录回调处理类
sasl.login.callback.handler.class=io.confluent.kafka.security.oauthbearer.OAuthBearerLoginCallbackHandler

三、验证认证是否生效的关键步骤 #

1. 服务端启动校验：
   • 查看Kafka Broker日志，正确配置下会显示成功连接Keycloak JWKS端点的日志；若clientId/Secret错误，会直接抛出认证失败异常，无法正常启动（你之前启动成功说明服务端未启用SASL认证，大概率还是用了PLAINTEXT监听）。
2. 无认证客户端测试：
   • 使用未配置任何SASL/OAuth2信息的生产者尝试生产，应收到Authentication failed错误，而非成功生产。
3. 错误凭证测试：
   • 配置错误的clientId/Secret到生产者，尝试生产时会收到Keycloak返回的invalid_client或invalid_grant错误，无法完成生产。
4. 手动令牌测试：
   • 调用Keycloak的token端点手动获取令牌，然后通过以下配置生产：

     # producer-with-token.properties
     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token="<你手动获取的令牌>";
     

     执行命令：

     kafka-console-producer.sh --broker-list <Kafka Broker地址>:9093 --topic test-topic --producer.config producer-with-token.properties
     

     若令牌过期或无效，生产会直接失败。

四、你之前问题的根因 #

你当前的Kafka服务端大概率仍在使用PLAINTEXT协议监听，导致SASL-OAuth2配置完全未生效，所以错误凭证不影响启动、无认证客户端也能正常生产。

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