Configure Confluent Server Brokers for SASL/OAUTHBEARER Authentication in Confluent Platform

To configure Confluent Server brokers for SASL/OAUTHBEARER authentication with TLS encryption (instead of the PLAINTEXT defaults), configure the properties file for each broker with the following property settings.

You can get the value of <idp-jwks-endpoint>, <idp-token-endpoint>, and others from the OIDC discovery endpoint. This is generally, https://<YOUR_IDP_DOMAIN>/.well-known/openid-configuration.

# Enable SASL/OAUTHBEARER mechanism
sasl.enabled.mechanisms=OAUTHBEARER

# Configure the OAUTHBEARER listener
listeners=SASL_SSL://:9092
advertised.listeners=SASL_SSL://hostname:9092
listener.name.sasl_ssl.sasl.enabled.mechanisms=OAUTHBEARER

# For RBAC setup, include the following to set the SASL callback validator and JAAS configuration:
listener.name.sasl_ssl.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.CompositeBearerValidatorCallbackHandler
listener.name.sasl_ssl.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule \
  required unsecuredLoginStringClaim_sub="thePrincipalName" publicKeyPath="</path/to/public.pem>";

# For non-RBAC setup, include the following to set the SASL callback validator and JAAS configuration:
listener.name.sasl_ssl.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler
listener.name.sasl_ssl.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;

# Configure the JWKS endpoint
listener.name.sasl_ssl.sasl.oauthbearer.jwks.endpoint.url=<https://your-jwks-endpoint.com/.well-known/jwks.json>

# Set expected audience and issuer
listener.name.sasl_ssl.sasl.oauthbearer.expected.audience=<your-audience>
listener.name.sasl_ssl.sasl.oauthbearer.expected.issuer=<https://your-issuer.com>

# Configure the principal builder
listener.name.sasl_ssl.sasl.oauthbearer.principal.builder.class=io.confluent.kafka.security.authenticator.OAuthKafkaPrincipalBuilder

# Set the groups claim name (if using group-based authorization)
listener.name.sasl_ssl.oauthbearer.groups.claim.name=groups

# Interbroker communication (optional)
sasl.mechanism.inter.broker.protocol=OAUTHBEARER
security.inter.broker.protocol=SASL_SSL

For details on the Confluent Server broker configuration properties used in this example, see Kafka Broker and Controller Configuration Reference for Confluent Platform.

Related content