Configure Kafka Connect for OAuth Authentication in Confluent Platform

This content provides a guide for configuring Kafka Connect with OAuth authentication, including connections to Metadata Service (MDS), REST endpoints, and connectors.

This guide details the configuration of Kafka Connect for OAuth authentication, covering essential steps for securing connections to the Metadata Service (MDS), REST endpoints, and connectors.

Configure Connect connection to MDS

To configure Connect to obtain an OAuth token from MDS, add the following configuration settings, replacing the placeholder values with your actual configuration values.

confluent.metadata.http.auth.credentials.provider=OAUTHBEARER
confluent.metadata.oauthbearer.token.endpoint.url=<url:port>
confluent.metadata.oauthbearer.login.client.id=<service-id>
confluent.metadata.oauthbearer.login.client.secret=<service-secret>

These include the minimal configurations required.

Configure optional properties

For the interaction between Connect and your identity provider (IdP), you can add the following optional configuration settings, which allow for customizing connection timeouts and retry mechanisms.

confluent.metadata.oauthbearer.login.connect.timeout.ms
confluent.metadata.oauthbearer.login.read.timeout.ms
confluent.metadata.oauthbearer.login.retry.backoff.max.ms
confluent.metadata.oauthbearer.login.retry.backoff.ms

Configure Connect REST endpoints

To configure the Connect REST endpoints with OAuth, add the following configuration settings for expected issuer and audience claims to ensure that REST API interactions are securely authenticated, replacing the placeholder values with your actual configuration values.

rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.AuthenticationHandler
oauthbearer.jwks.endpoint.url=<jwks-endpoint>
oauthbearer.expected.issuer=<expected-issuer>
oauthbearer.expected.audience=<expected-audience>
oauthbearer.sub.claim.name=<sub-claim-name>
oauthbearer.groups.claim.name=<groups-claim-name>

The AuthenticationHandler class enables OAuth authentication on Connect REST endpoints, managing the authentication of all requests. For more information, see Use the AuthenticationHandler Class for Multi-Protocol Authentication in Confluent Platform.

Configure connectors as sink or source

Connectors can act as producers, consumers, or admin Kafka clients. To enable this, configure SASL/OAUTHBEARER settings as outlined in KIP-768. These configuration properties can be applied at:

  • Connect Worker level: Allows all connectors to share the same ID and secret.
  • Connector level: Enables specific role bindings for each connector.

Configure Connect Worker level configurations for connectors

Add the following configurations to enable OAuth authentication for Kafka Connect workers, allowing them to securely produce and consume messages using the SASL_SSL protocol. By specifying the OAUTHBEARER mechanism, these settings ensure that both producers and consumers authenticate using OAuth tokens, leveraging the OAuthBearerLoginCallbackHandler for token management. The use of SASL_SSL ensures that data in transit is encrypted, enhancing the security of your Kafka Connect deployment. Replace the placeholder values with your actual configuration values.

producer.security.protocol=SASL_SSL
producer.sasl.mechanism=OAUTHBEARER
producer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler
producer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    clientId="<client-id>" \
    clientSecret="<client-secret>" \
    scope="<scope>";

consumer.security.protocol=SASL_SSL
consumer.sasl.mechanism=OAUTHBEARER
consumer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler
consumer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    clientId="<client-id>" \
    clientSecret="<client-secret>" \
    scope="<scope>";

Configure connector level properties

Use the following configurations to enforce specific role bindings for each connector. Replace the placeholder values with your actual configuration values to ensure secure and authenticated communication.

producer.override.security.protocol=SASL_SSL
producer.override.sasl.mechanism=OAUTHBEARER
producer.override.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler
producer.override.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    clientId="<client-id>" \
    clientSecret="<client-secret>" \
    scope="<scope>";

Related content