OAuth Configuration Reference for Confluent Cloud Clients

This section provides a comprehensive reference for OAuth configuration parameters across different client types and platforms.

Java client parameters

Parameter Description Default Required
sasl.mechanism SASL mechanism for OAuth authentication OAUTHBEARER Yes
sasl.jaas.config JAAS configuration for OAuth login module None Yes
sasl.login.callback.handler.class OAuth callback handler class io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler Yes
sasl.login.connect.timeout.ms Connection timeout for token requests 10000 No
sasl.login.read.timeout.ms Read timeout for token requests 10000 No
sasl.login.retry.backoff.ms Initial retry backoff for failed requests 100 No
sasl.login.retry.backoff.max.ms Maximum retry backoff for failed requests 10000 No
oauth.token.endpoint.uri OAuth token endpoint URL None Yes
oauth.client.id OAuth client ID None Yes
oauth.client.secret OAuth client secret None Yes
oauth.scope OAuth scopes for token request None No
oauth.logical.cluster Logical cluster ID for Confluent Cloud. Required except when connecting to AWS Enterprise clusters over AWS PrivateLink. Can be omitted due to server-side mapping. None Conditional
oauth.identity.pool.id Identity pool ID for Confluent Cloud None No

JAAS configuration parameters

Parameter Description Default Required
oauth.token.endpoint.uri OAuth token endpoint URL None Yes
oauth.client.id OAuth client ID None Yes
oauth.client.secret OAuth client secret None Yes
oauth.scope OAuth scopes for token request None No
oauth.logical.cluster Logical cluster ID for Confluent Cloud. Required except when connecting to AWS Enterprise clusters over AWS PrivateLink; may be omitted due to server-side mapping. None Conditional
oauth.identity.pool.id Identity pool ID for Confluent Cloud None No

Python client parameters

Parameter Description Default Required
sasl.mechanism SASL mechanism for OAuth authentication OAUTHBEARER Yes
sasl.oauth.token.endpoint.uri OAuth token endpoint URL None Yes
sasl.oauth.client.id OAuth client ID None Yes
sasl.oauth.client.secret OAuth client secret None Yes
sasl.oauth.scope OAuth scopes for token request None No
sasl.oauth.logical.cluster Logical cluster ID for Confluent Cloud. Required except when connecting to AWS Enterprise clusters over AWS PrivateLink; may be omitted due to server-side mapping. None Conditional
sasl.oauth.identity.pool.id Identity pool ID for Confluent Cloud None No
sasl.oauth.connect.timeout.ms Connection timeout for token requests 10000 No
sasl.oauth.read.timeout.ms Read timeout for token requests 10000 No

.NET client parameters

Parameter Description Default Required
sasl.mechanism SASL mechanism for OAuth authentication OAUTHBEARER Yes
sasl.oauth.token.endpoint.uri OAuth token endpoint URL None Yes
sasl.oauth.client.id OAuth client ID None Yes
sasl.oauth.client.secret OAuth client secret None Yes
sasl.oauth.scope OAuth scopes for token request None No
sasl.oauth.logical.cluster Logical cluster ID for Confluent Cloud. Required except when connecting to AWS Enterprise clusters over AWS PrivateLink; may be omitted due to server-side mapping. None Conditional
sasl.oauth.identity.pool.id Identity pool ID for Confluent Cloud None No
sasl.oauth.connect.timeout.ms Connection timeout for token requests 10000 No
sasl.oauth.read.timeout.ms Read timeout for token requests 10000 No

Go client parameters

Parameter Description Default Required
sasl.mechanism SASL mechanism for OAuth authentication OAUTHBEARER Yes
sasl.oauth.token.endpoint.uri OAuth token endpoint URL None Yes
sasl.oauth.client.id OAuth client ID None Yes
sasl.oauth.client.secret OAuth client secret None Yes
sasl.oauth.scope OAuth scopes for token request None No
sasl.oauth.logical.cluster Logical cluster ID for Confluent Cloud. Required except when connecting to AWS Enterprise clusters over AWS PrivateLink; may be omitted due to server-side mapping. None Conditional
sasl.oauth.identity.pool.id Identity pool ID for Confluent Cloud None No
sasl.oauth.connect.timeout.ms Connection timeout for token requests 10000 No
sasl.oauth.read.timeout.ms Read timeout for token requests 10000 No

Schema Registry client parameters

Parameter Description Default Required
bearer.auth.credentials.source Authentication method. Use OAUTHBEARER for OAuth None Yes
bearer.auth.issuer.endpoint.url OAuth token endpoint URL None Yes
bearer.auth.client.id OAuth client ID None Yes
bearer.auth.client.secret OAuth client secret None Yes
bearer.auth.scope OAuth scopes for token request None No
bearer.auth.logical.cluster Schema Registry logical cluster ID (lsrc-xxxxx) None Yes
bearer.auth.identity.pool.id Identity pool ID for Confluent Cloud None No

Configuration validation

When configuring OAuth authentication, validate your configuration:

  • Required parameters: Ensure all required parameters are provided.
  • URL format: Verify token endpoint URLs are properly formatted
  • Credentials: Validate client ID and secret are correct
  • Network access: Confirm network connectivity to OAuth provider
  • Permissions: Verify client has appropriate scopes and permissions
// Configuration validation example
public static void validateOAuthConfig(Properties props) {
    // Check required parameters
    String[] required = {
        "sasl.mechanism",
        "sasl.jaas.config",
        "oauth.token.endpoint.uri",
        "oauth.client.id",
        "oauth.client.secret",
        "oauth.logical.cluster"
    };

    for (String param : required) {
        if (!props.containsKey(param)) {
            throw new IllegalArgumentException("Missing required parameter: " + param);
        }
    }

    // Validate timeout values
    int connectTimeout = Integer.parseInt(props.getProperty("sasl.login.connect.timeout.ms", "10000"));
    int readTimeout = Integer.parseInt(props.getProperty("sasl.login.read.timeout.ms", "10000"));

    if (connectTimeout < 1000 || readTimeout < 1000) {
        throw new IllegalArgumentException("Timeout values must be at least 1000ms");
    }
}

Configuration templates

Use configuration templates for consistent setup:

# oauth-config-template.yaml
kafka:
  oauth:
    token_endpoint: ${OAUTH_TOKEN_ENDPOINT}
    client_id: ${OAUTH_CLIENT_ID}
    client_secret: ${OAUTH_CLIENT_SECRET}
    scope: ${OAUTH_SCOPE}
    logical_cluster: ${LOGICAL_CLUSTER_ID}
    identity_pool: ${IDENTITY_POOL_ID}
    timeouts:
      connect: 10000
      read: 10000
      retry_backoff: 100
      retry_max: 10000

Environment-specific configurations

Development environment

# Development OAuth configuration
sasl.oauthbearer.token.endpoint.url=https://dev-auth.example.com/oauth2/token
sasl.oauthbearer.client.id=dev-kafka-client
sasl.oauthbearer.client.secret=${DEV_CLIENT_SECRET}
sasl.oauthbearer.scope=kafka:dev

Staging environment

# Staging OAuth configuration
sasl.oauthbearer.token.endpoint.url=https://staging-auth.example.com/oauth2/token
sasl.oauthbearer.client.id=staging-kafka-client
sasl.oauthbearer.client.secret=${STAGING_CLIENT_SECRET}
sasl.oauthbearer.scope=kafka:staging

Production environment

# Production OAuth configuration
sasl.oauthbearer.token.endpoint.url=https://prod-auth.example.com/oauth2/token
sasl.oauthbearer.client.id=prod-kafka-client
sasl.oauthbearer.client.secret=${PROD_CLIENT_SECRET}
sasl.oauthbearer.scope=kafka:prod

Advanced configuration options

Custom token providers

For advanced use cases, implement custom token providers:

public class CustomTokenProvider implements BearerAuthCredentialProvider {

    private final String tokenEndpoint;
    private final String clientId;
    private final String clientSecret;
    private final Map<String, String> additionalHeaders;

    public CustomTokenProvider(String tokenEndpoint, String clientId,
                             String clientSecret, Map<String, String> additionalHeaders) {
        this.tokenEndpoint = tokenEndpoint;
        this.clientId = clientId;
        this.clientSecret = clientSecret;
        this.additionalHeaders = additionalHeaders;
    }

    @Override
    public String getBearerToken(URL url) throws IOException {
        // Custom token retrieval logic
        HttpURLConnection connection = (HttpURLConnection) new URL(tokenEndpoint).openConnection();
        connection.setRequestMethod("POST");
        connection.setRequestProperty("Content-Type", "application/x-www-form-urlencoded");

        // Add custom headers
        if (additionalHeaders != null) {
            for (Map.Entry<String, String> header : additionalHeaders.entrySet()) {
                connection.setRequestProperty(header.getKey(), header.getValue());
            }
        }

        // Set up authentication
        String auth = clientId + ":" + clientSecret;
        String encodedAuth = Base64.getEncoder().encodeToString(auth.getBytes());
        connection.setRequestProperty("Authorization", "Basic " + encodedAuth);

        // Send request
        String postData = "grant_type=client_credentials";
        connection.setDoOutput(true);
        try (OutputStream os = connection.getOutputStream()) {
            os.write(postData.getBytes());
        }

        // Parse response
        try (BufferedReader br = new BufferedReader(
                new InputStreamReader(connection.getInputStream()))) {
            StringBuilder response = new StringBuilder();
            String line;
            while ((line = br.readLine()) != null) {
                response.append(line);
            }

            // Parse JSON response
            ObjectMapper mapper = new ObjectMapper();
            JsonNode jsonNode = mapper.readTree(response.toString());
            return jsonNode.get("access_token").asText();
        }
    }
}

Retry and circuit breaker patterns

public class CircuitBreaker {
    private final AtomicInteger failureCount = new AtomicInteger(0);
    private final AtomicLong lastFailureTime = new AtomicLong(0);
    private final int threshold;
    private final long timeout;

    public CircuitBreaker(int threshold, long timeout) {
        this.threshold = threshold;
        this.timeout = timeout;
    }

    public boolean isOpen() {
        long now = System.currentTimeMillis();
        if (failureCount.get() >= threshold) {
            if (now - lastFailureTime.get() < timeout) {
                return true; // Circuit is open
            } else {
                // Reset after timeout
                failureCount.set(0);
                return false;
            }
        }
        return false;
    }

    public void recordFailure() {
        failureCount.incrementAndGet();
        lastFailureTime.set(System.currentTimeMillis());
    }

    public void recordSuccess() {
        failureCount.set(0);
    }
}

public class RetryTemplate {
    private final int maxRetries;
    private final long backoffMs;

    public RetryTemplate(int maxRetries, long backoffMs) {
        this.maxRetries = maxRetries;
        this.backoffMs = backoffMs;
    }

    public <T> T execute(Supplier<T> operation) throws Exception {
        int attempts = 0;
        while (attempts < maxRetries) {
            try {
                return operation.get();
            } catch (Exception e) {
                attempts++;
                if (attempts >= maxRetries) {
                    throw e;
                }
                Thread.sleep(backoffMs * attempts);
            }
        }
        throw new RuntimeException("Max retries exceeded");
    }
}

Deployment considerations

Container deployment

# Dockerfile for OAuth-enabled Kafka client
FROM openjdk:11-jre-slim

# Copy application
COPY target/oauth-kafka-client.jar /app/

# Set environment variables
ENV OAUTH_TOKEN_ENDPOINT=https://your-oauth-provider.com/oauth2/token
ENV OAUTH_CLIENT_ID=your-client-id
ENV OAUTH_CLIENT_SECRET=your-client-secret
ENV OAUTH_SCOPE=kafka:read kafka:write
ENV LOGICAL_CLUSTER_ID=lkc-xxxxx
ENV IDENTITY_POOL_ID=pool-xxxxx

# Run application
CMD ["java", "-jar", "/app/oauth-kafka-client.jar"]

Kubernetes deployment

# kubernetes-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: oauth-kafka-client
spec:
  replicas: 3
  selector:
    matchLabels:
      app: oauth-kafka-client
  template:
    metadata:
      labels:
        app: oauth-kafka-client
    spec:
      containers:
      - name: kafka-client
        image: your-registry/oauth-kafka-client:latest
        env:
        - name: OAUTH_TOKEN_ENDPOINT
          valueFrom:
            secretKeyRef:
              name: oauth-secrets
              key: token-endpoint
        - name: OAUTH_CLIENT_ID
          valueFrom:
            secretKeyRef:
              name: oauth-secrets
              key: client-id
        - name: OAUTH_CLIENT_SECRET
          valueFrom:
            secretKeyRef:
              name: oauth-secrets
              key: client-secret
        - name: OAUTH_SCOPE
          value: "kafka:read kafka:write"
        - name: LOGICAL_CLUSTER_ID
          value: "lkc-xxxxx"
        - name: IDENTITY_POOL_ID
          value: "pool-xxxxx"

CI/CD integration

# .github/workflows/deploy.yml
name: Deploy OAuth Kafka Client

on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2

    - name: Set up JDK
      uses: actions/setup-java@v2
      with:
        java-version: '11'

    - name: Build with Maven
      run: mvn clean package

    - name: Build Docker image
      run: docker build -t oauth-kafka-client .

    - name: Deploy to Kubernetes
      run: |
        kubectl apply -f kubernetes-deployment.yaml
        kubectl rollout restart deployment/oauth-kafka-client

Integration testing

Mock OAuth server

@Test
public void testOAuthIntegration() {
    // Start mock OAuth server
    MockOAuthServer mockServer = new MockOAuthServer();
    mockServer.start();

    try {
        // Configure client with mock server
        Properties props = new Properties();
        props.put("sasl.mechanism", "OAUTHBEARER");
        props.put("sasl.jaas.config",
            "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required " +
            "oauth.token.endpoint.uri=\"" + mockServer.getTokenEndpoint() + "\" " +
            "oauth.client.id=\"test-client\" " +
            "oauth.client.secret=\"test-secret\";");

        // Test authentication
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        // ... test operations

    } finally {
        mockServer.stop();
    }
}

Performance testing

@Test
public void testOAuthPerformance() {
    // Measure token retrieval performance
    long startTime = System.currentTimeMillis();

    for (int i = 0; i < 100; i++) {
        // Perform OAuth token request
        String token = oauthProvider.getToken();
        assertNotNull(token);
    }

    long endTime = System.currentTimeMillis();
    long duration = endTime - startTime;

    // Assert performance requirements
    assertTrue("OAuth performance below threshold", duration < 5000);
}

Azure User-Assigned Managed Identity (UAMI)

Confluent Cloud supports using Azure User-Assigned Managed Identity (UAMI) for OAuth authentication, eliminating the need for static client IDs and secrets. This feature leverages Azure’s built-in identity management to automatically retrieve authentication tokens.

For complete documentation on configuring and using UAMI with OAuth, including step-by-step setup instructions and client examples, see Configure Azure User Assigned Managed Identity OAuth for Confluent Cloud.

Related content