# Configure Kafka Connect for OAuth Authentication in Confluent Platform 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. ```properties confluent.metadata.http.auth.credentials.provider=OAUTHBEARER confluent.metadata.oauthbearer.token.endpoint.url= confluent.metadata.oauthbearer.login.client.id= confluent.metadata.oauthbearer.login.client.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. ```properties 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 ``` For Kafka Java clients supporting SASL OAUTHBEARER, allow specific IdP endpoints by setting the following configuration property: ```properties org.apache.kafka.sasl.oauthbearer.allowed.urls=,,... ``` This property specifies a comma-separated list of allowed IdP JWKS (JSON Web Key Set) and token endpoint URLs. Use \* (asterisk) as the value to allow any endpoint. ```properties org.apache.kafka.sasl.oauthbearer.allowed.urls=* ``` You should consult the specific Kafka client and IdP documentation for the exact interpretation and security implications of such a broad setting. Java applications should set this property as a JVM system property when launching the application: ```bash -Dorg.apache.kafka.sasl.oauthbearer.allowed.urls=,,... ``` For other clients (for example, Python, Go, .NET) that are built on librdkafka, these clients use different property names and configuration mechanisms. So, refer to specific client library documentation for the equivalent OAuthBEARER configuration properties. ## 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. ```properties rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.AuthenticationHandler oauthbearer.jwks.endpoint.url= oauthbearer.expected.issuer= oauthbearer.expected.audience= oauthbearer.sub.claim.name= oauthbearer.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](../multi-protocol/authenticationhandler.md#authenticationhandler). ## 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](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=186877575). 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, using 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. ```properties producer.security.protocol=SASL_SSL producer.sasl.mechanism=OAUTHBEARER producer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler producer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \ clientId="" \ clientSecret="" \ scope=""; consumer.security.protocol=SASL_SSL consumer.sasl.mechanism=OAUTHBEARER consumer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler consumer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \ clientId="" \ clientSecret="" \ 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. ```properties producer.override.security.protocol=SASL_SSL producer.override.sasl.mechanism=OAUTHBEARER producer.override.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler producer.override.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \ clientId="" \ clientSecret="" \ scope=""; ``` ## Related content * [Configure Truststores for TLS Handshake with Identity Providers](configure-truststore.md#configure-truststore-for-oauth) * [Use OAuth/OIDC for Authentication in Confluent Platform](overview.md#oauth-oidc-authentication-overview) * [Use SASL/OAUTHBEARER Authentication Between Confluent Server Brokers and Kafka Clients in Confluent Platform](../sasl/oauthbearer/overview.md#sasl-oauthbearer-authentication-overview)