Kafka 3.9.0集成Okta OAuthBearer启动报错：JWT sub无主体名

错误信息 #

[2024-12-16 23:31:44,482] ERROR No principal name in JWT claim: sub (org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule)
java.io.IOException: No principal name in JWT claim: sub
    at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler.handle(OAuthBearerUnsecuredLoginCallbackHandler.java:166)
    at org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.identifyToken(OAuthBearerLoginModule.java:317)
    at org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.login(OAuthBearerLoginModule.java:302)
    at java.base/javax.security.auth.login.LoginContext.invoke(LoginContext.java:755)
    at java.base/javax.security.auth.login.LoginContext$4.run(LoginContext.java:679)
    at java.base/javax.security.auth.login.LoginContext$4.run(LoginContext.java:677)
    at java.base/java.security.AccessController.doPrivileged(AccessController.java:712)
    at java.base/javax.security.auth.login.LoginContext.invokePriv(LoginContext.java:677)
    at java.base/javax.security.auth.login.LoginContext.login(LoginContext.java:587)
    at org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin.login(ExpiringCredentialRefreshingLogin.java:205)
    at org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerRefreshingLogin.login(OAuthBearerRefreshingLogin.java:151)
    at org.apache.kafka.common.security.authenticator.LoginManager.<init>(LoginManager.java:64)
    at org.apache.kafka.common.security.authenticator.LoginManager.acquireLoginManager(LoginManager.java:107)
    at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:173)
    at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:193)
    at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:82)
    at kafka.server.NodeToControllerChannelManagerImpl.buildNetworkClient$1(NodeToControllerChannelManager.scala:162)
    at kafka.server.NodeToControllerChannelManagerImpl.newRequestThread(NodeToControllerChannelManager.scala:202)
    at kafka.server.NodeToControllerChannelManagerImpl.<init>(NodeToControllerChannelManager.scala:141)
    at kafka.server.KafkaServer.startup(KafkaServer.scala:357)
    at kafka.Kafka$.main(Kafka.scala:112)
    at kafka.Kafka.main(Kafka.scala)
Caused by: org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerConfigException: No principal name in JWT claim: sub
    at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler.handleTokenCallback(OAuthBearerUnsecuredLoginCallbackHandler.java:220)
    at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler.handle(OAuthBearerUnsecuredLoginCallbackHandler.java:164)
    ... 21 more
Caused by: org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerIllegalTokenException: No principal name in JWT claim: sub
    at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredJws.<init>(OAuthBearerUnsecuredJws.java:109)
    at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler.handleTokenCallback(OAuthBearerUnsecuredLoginCallbackHandler.java:212)
    ... 22 more

当前server.properties配置 #

listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://localhost:9092
inter.broker.listener.name=SASL_PLAINTEXT
listener.security.protocol.map=SASL_PLAINTEXT:SASL_PLAINTEXT

# SASL and OAuth2 Configuration
sasl.mechanism.inter.broker.protocol=OAUTHBEARER
listener.name.sasl_plaintext.sasl.enabled.mechanisms=OAUTHBEARER
listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler

# JWKS endpoint for token validation
sasl.oauthbearer.jwks.endpoint.url=https://dev-someId.okta.com/oauth2/default/v1/keys


# JAAS Configuration for OAuth2 (Inline in server.properties for simplicity)
listener.name.sasl_plaintext.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    clientId="xxxxxxxxxx" \
    clientSecret="xxxxxxxxxvxvxvxvxvvxvxvxvxvvxvxvxvxvxvxvxv" \
    scope="kafka" \
    grantType="client_credentials" \
    tokenEndpointUrl="https://dev-someid.okta.com/oauth2/default/v1/token";

解码后的JWT Token内容 #

{
  "ver": 1,
  "jti": "AT.XxXY_2aH9Oxz5IC-QruS4Iqhh_7TrfCM6Sj3lzDnKDY",
  "iss": "https://dev-someId.okta.com/oauth2/default",
  "aud": "api://default",
  "iat": 1734369931,
  "exp": 1734373531,
  "cid": "0oalmwzen2tCuDxB05d7",
  "scp": [
    "kafka"
  ],
  "sub": "0oalmwzen2tCuDxB05d7"
}

错误原因分析 #

错误提示指向JWT的sub字段无法提取有效主体名称，尽管Token中sub字段存在且有值，但Kafka默认的OAuthBearerLoginModule在**客户端凭证模式（client_credentials）**下，对sub字段的解析逻辑与Okta返回格式不兼容：

• Okta在客户端凭证模式下，sub字段返回的是客户端ID（与cid字段值一致）
• Kafka默认期望sub字段代表用户主体，而非客户端ID，导致解析时判定为无效主体

解决方案 #

在JAAS配置的OAuthBearerLoginModule中添加principalClaimName参数，指定使用cid（客户端ID）作为主体来源，调整后的JAAS配置如下：

listener.name.sasl_plaintext.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    clientId="xxxxxxxxxx" \
    clientSecret="xxxxxxxxxvxvxvxvxvvxvxvxvxvvxvxvxvxvxvxvxv" \
    scope="kafka" \
    grantType="client_credentials" \
    tokenEndpointUrl="https://dev-someid.okta.com/oauth2/default/v1/token" \
    principalClaimName="cid";

额外验证项 #

若仍报错，需确认：

• Okta的Token端点返回的sub字段为非空字符串（当前Token已满足）
• Kafka服务器可正常访问Okta的JWKS和Token端点
• 客户端ID、密钥及权限范围配置正确

Kafka与Okta OAUTHBEARER标准配置参考 #

以下是适配Okta的Kafka Broker标准配置模板：

# 基础监听器配置
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://localhost:9092
inter.broker.listener.name=SASL_PLAINTEXT
listener.security.protocol.map=SASL_PLAINTEXT:SASL_PLAINTEXT

# SASL OAuth2核心配置
sasl.mechanism.inter.broker.protocol=OAUTHBEARER
listener.name.sasl_plaintext.sasl.enabled.mechanisms=OAUTHBEARER

# 服务器端Token验证回调处理器
listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler

# Okta JWKS端点（用于Token签名验证）
sasl.oauthbearer.jwks.endpoint.url=https://dev-someId.okta.com/oauth2/default/v1/keys

# 客户端凭证模式JAAS配置
listener.name.sasl_plaintext.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    clientId="YOUR_OKTA_CLIENT_ID" \
    clientSecret="YOUR_OKTA_CLIENT_SECRET" \
    scope="kafka" \
    grantType="client_credentials" \
    tokenEndpointUrl="https://dev-someId.okta.com/oauth2/default/v1/token" \
    principalClaimName="cid";

# 可选：验证Token受众（需与Okta配置的aud一致）
# sasl.oauthbearer.expected.audience=api://default

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