# 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 or Dedicated clusters over AWS PrivateLink, where it can be omitted due to server-side mapping. For clusters using public networking, this parameter is required. | 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 or Dedicated clusters over AWS PrivateLink, where it can be omitted due to server-side mapping. For clusters using public networking, this parameter is required. | 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.oauthbearer.token.endpoint.url` | OAuth token endpoint URL | None | Yes |
| `sasl.oauthbearer.client.id` | OAuth client ID | None | Yes |
| `sasl.oauthbearer.client.secret` | OAuth client secret | None | Yes |
| `sasl.oauthbearer.scope` | OAuth scopes for token request | None | No |
## .NET client parameters
| Parameter | Description | Default | Required |
|---------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------|-------------|
| `sasl.mechanism` | SASL mechanism for OAuth authentication | OAUTHBEARER | Yes |
| `sasl.oauthbearer.token.endpoint.url` | OAuth token endpoint URL | None | Yes |
| `sasl.oauthbearer.client.id` | OAuth client ID | None | Yes |
| `sasl.oauthbearer.client.secret` | OAuth client secret | None | Yes |
| `sasl.oauthbearer.scope` | OAuth scopes for token request | None | No |
| `sasl.oauthbearer.logical.cluster` | Logical cluster ID for Confluent Cloud. Required except when connecting to AWS Enterprise or Dedicated clusters over AWS PrivateLink, where it can be omitted due to server-side mapping. For clusters using public networking, this parameter is required. | None | Conditional |
| `sasl.oauthbearer.identity.pool.id` | Identity pool ID for Confluent Cloud | None | No |
| `sasl.oauthbearer.connect.timeout.ms` | Connection timeout for token requests | 10000 | No |
| `sasl.oauthbearer.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.oauthbearer.token.endpoint.url` | OAuth token endpoint URL | None | Yes |
| `sasl.oauthbearer.client.id` | OAuth client ID | None | Yes |
| `sasl.oauthbearer.client.secret` | OAuth client secret | None | Yes |
| `sasl.oauthbearer.scope` | OAuth scopes for token request | None | No |
| `sasl.oauthbearer.logical.cluster` | Logical cluster ID for Confluent Cloud. Required except when connecting to AWS Enterprise or Dedicated clusters over AWS PrivateLink, where it can be omitted due to server-side mapping. For clusters using public networking, this parameter is required. | None | Conditional |
| `sasl.oauthbearer.identity.pool.id` | Identity pool ID for Confluent Cloud | None | No |
| `sasl.oauthbearer.connect.timeout.ms` | Connection timeout for token requests | 10000 | No |
| `sasl.oauthbearer.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
Accepts a comma-separated list of pool
IDs. If omitted, auto pool mapping applies.
`confluent-kafka` version 2.15.0 or
later is required to omit this parameter
or to specify a comma-separated list of pool IDs.
For more information, see
[Use auto pool mapping with OAuth identity pools](../identity-pools.md#oauth-auto-pool-mapping). | 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
```java
// 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:
```yaml
# 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
```properties
# 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
```properties
# 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
```properties
# 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:
```java
public class CustomTokenProvider implements BearerAuthCredentialProvider {
private final String tokenEndpoint;
private final String clientId;
private final String clientSecret;
private final Map additionalHeaders;
public CustomTokenProvider(String tokenEndpoint, String clientId,
String clientSecret, Map 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 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
```java
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 execute(Supplier 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
# 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
```yaml
# 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
```yaml
# .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
```java
@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 producer = new KafkaProducer<>(props);
// ... test operations
} finally {
mockServer.stop();
}
}
```
#### Performance testing
```java
@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](uami.md#oauth-uami-share).
## Related content
- [Configure Azure User Assigned Managed Identity OAuth for Confluent Cloud](uami.md#oauth-uami-share)
- [Configure Kafka Clients for OAuth 2.0 Authentication in Confluent Cloud](overview.md#oauth-client-configuration-overview)
- [Use OAuth/OIDC to Authenticate to Confluent Cloud](../overview.md#oauth-overview)
- [Configure Java Clients for OAuth/OIDC on Confluent Cloud](java-clients.md#configure-java-clients-oauth)
- [Configure Python Clients for OAuth/OIDC on Confluent Cloud](python-clients.md#configure-python-clients-oauth)
- [Configure .NET Clients for OAuth/OIDC on Confluent Cloud](dotnet-clients.md#configure-dotnet-clients-oauth)
- [Configure Go Clients for OAuth/OIDC on Confluent Cloud](go-clients.md#configure-go-clients-oauth)
- [Configure JavaScript Clients for OAuth/OIDC on Confluent Cloud](javascript-clients.md#configure-javascript-clients-oauth)
- [Troubleshoot OAuth/OIDC Issues on Confluent Cloud](../troubleshooting.md#oauth-troubleshooting)