Configure AWS IAM OAuth for Java Client

Confluent Cloud users running Apache Kafka® clients on AWS can authenticate with AWS IAM identities instead of static OAuth credentials. The plugin retrieves a signed JSON Web Token (JWT) from AWS Security Token Service (STS) using the GetWebIdentityToken API. Confluent Cloud validates the JWT against AWS’s published JWKS endpoint, removing the need to manage and rotate client secrets.

The plugin is delivered as a Confluent-supported Java plugin in the kafka-client-plugins Maven artifact. AWS IAM policies do not govern Confluent Cloud authorization. Authorization is handled by Confluent Cloud RBAC bound to an identity pool. The plugin’s scope is identity only.

Benefits of using AWS IAM OAuth

Enhanced security

Eliminates the need to store and manage static client secrets
Reduces the risk of credential exposure and theft
Uses AWS’s managed identity infrastructure

Simplified credential management

No need to manually rotate client secrets
AWS manages the credential lifecycle automatically
Reduces operational overhead and compliance burden

AWS integration

Native integration with AWS resources and services
Works with existing AWS IAM roles and policies
Supports cloud-native security best practices

Prerequisites

AWS IAM OAuth requires an AWS account, AWS Outbound Identity Federation enabled, and a configured Confluent Cloud identity pool. Complete the following prerequisites:

Your environment must use an AWS account that supports IAM roles, such as EC2, Lambda, EKS, ECS, or Fargate. Your Kafka client can run inside an AWS compute environment or another environment if you use the AWS default credentials chain. For more about default credentials, see AWS documentation: Default credentials provider chain.
AWS Outbound Identity Federation is enabled on your AWS account.
An IAM policy that grants sts:GetWebIdentityToken is attached to the IAM role used by your AWS compute resource.
A Confluent Cloud OAuth identity provider is configured to trust your AWS account’s issuer URL and reference AWS’s JWKS URI.
A Confluent Cloud identity pool is configured and mapped to the Confluent Cloud RBAC role bindings your client needs.
Java Client: version 8.3 or later of kafka-client-plugins.

Configure Java Client

To configure Java Client for AWS IAM OAuth, add the Confluent Maven repository and the kafka-client-plugins dependency to your project, then configure the client properties. The AWS IAM OAuth Java Client plugin is available in Confluent Platform 8.3 and later, distributed in the Confluent kafka-client-plugins Maven artifact.

Add the plugin dependency

Add the Confluent Maven repository and the plugin dependency to your project.

Maven

<repositories>
  <repository>
    <id>confluent</id>
    <url>https://packages.confluent.io/maven/</url>
  </repository>
</repositories>

<dependencies>
  <dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-client-plugins</artifactId>
    <version><VERSION></version>
  </dependency>
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version><VERSION></version>
  </dependency>
</dependencies>

Gradle

repositories {
    maven {
        url "https://packages.confluent.io/maven/"
    }
}

dependencies {
    implementation 'io.confluent:kafka-client-plugins:<VERSION>'
    implementation 'org.apache.kafka:kafka-clients:<VERSION>'
}

Configure client properties

Configure your Java client with the following properties. Replace placeholders with your actual values.

# Connection
bootstrap.servers=<CONFLUENT_CLOUD_BOOTSTRAP>

# Serializers
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer

# OAuth using AWS STS WebIdentityToken
security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
sasl.oauthbearer.jwt.retriever.class=io.confluent.security.auth.client.oauth.AwsIamJwtRetriever
sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler

# AWS STS WebIdentityToken parameters
sasl.oauthbearer.aws.iam.webidentitytoken.audience=<CONFLUENT_CLOUD_BOOTSTRAP>
sasl.oauthbearer.aws.iam.webidentitytoken.signing.algorithm=ES384
sasl.oauthbearer.aws.iam.webidentitytoken.duration.seconds=300
sasl.oauthbearer.aws.iam.webidentitytoken.region=<AWS_REGION>
sasl.oauthbearer.aws.iam.webidentitytoken.tags=team=<TEAM>,environment=<ENVIRONMENT>

# JAAS — Confluent Cloud identity pool binding
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    clientId='<CLIENT_ID>' \
    scope='kafka' \
    extension_logicalCluster='<LOGICAL_CLUSTER>' \
    extension_identityPoolId='<IDENTITY_POOL_ID>';

Parameter reference

Property	Required	Description
`sasl.oauthbearer.jwt.retriever.class`	Yes	Set to `io.confluent.security.auth.client.oauth.AwsIamJwtRetriever` to use the AWS IAM JWT retriever.
`sasl.login.callback.handler.class`	Yes	Set to `org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler` (the standard Kafka OAuthBearer callback).
`sasl.oauthbearer.aws.iam.webidentitytoken.audience`	Yes	The audience claim embedded in the JWT issued by AWS STS. Set to your Confluent Cloud cluster bootstrap URL, including port. Must match the audience configured on your Confluent Cloud OAuth identity provider.
`sasl.oauthbearer.aws.iam.webidentitytoken.signing.algorithm`	Yes	JWT signing algorithm. Supported values: `RS256`, `ES384`.
`sasl.oauthbearer.aws.iam.webidentitytoken.duration.seconds`	Optional	Lifetime of the requested token in seconds. Default: `300`. Valid range: 60 to 3,600. The maximum is also bounded by the IAM role’s max session duration.
`sasl.oauthbearer.aws.iam.webidentitytoken.region`	Optional	AWS region for the STS endpoint, for example `us-west-2`. If unset, the plugin uses the AWS SDK default region resolution chain (environment variables, profile, instance metadata).
`sasl.oauthbearer.aws.iam.webidentitytoken.tags`	Optional	Comma-separated `key=value` tags attached to the STS session. Use to meet AWS resource-tagging policies.
`sasl.login.retry.backoff.ms`	Optional	Initial backoff for retrying failed STS calls. If unset, uses the standard Kafka client default.
`sasl.login.retry.backoff.max.ms`	Optional	Maximum backoff for retrying failed STS calls. If unset, uses the standard Kafka client default.
JAAS `clientId`	Yes	An identifier for the calling client; surfaces in audit logs.
JAAS `scope`	Yes	Set to `kafka`.
JAAS `extension_logicalCluster`	Yes	Your Confluent Cloud Kafka cluster ID, for example `lkc-xxxxxx`.
JAAS `extension_identityPoolId`	Yes	The Confluent Cloud identity pool ID this client authenticates through, for example `pool-xxxxxx`.

Important

The audience value must exactly match what your Confluent Cloud OAuth identity provider is configured to accept. If authentication fails with an audience-mismatch error, check the identity provider configuration in the Confluent Cloud Console.

Verify your client

To verify your configuration, run a test producer that writes timestamped messages to a topic and confirm they appear in the console.

Create a topic in your cluster, for example aws-iam-test.

Build and run a sample producer that reads the properties file from Configure client properties and sends 10 messages. Each message uses the current ISO 8601 timestamp as the key and message-N as the value. The producer prints the assigned partition and offset for each message.

A minimal Java producer:

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.io.FileInputStream;
import java.time.Instant;
import java.util.Properties;

public class AwsIamOAuthProducer {
    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        try (FileInputStream in = new FileInputStream(args[0])) {
            props.load(in);
        }
        String topic = props.getProperty("topic", "aws-iam-test");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            for (int i = 0; i < 10; i++) {
                String key = Instant.now().toString();
                String value = "message-" + i;
                RecordMetadata md = producer.send(new ProducerRecord<>(topic, key, value)).get();
                System.out.printf("Sent: key=%s value=%s -> partition=%d offset=%d%n",
                    key, value, md.partition(), md.offset());
            }
        }
        System.out.println("Done.");
    }
}

Confirm the program prints 10 Sent: lines followed by Done.. For example:

Sent: key=2026-04-20T22:20:16.053565936Z value=message-0 -> partition=0 offset=0
Sent: key=2026-04-20T22:20:17.910871467Z value=message-1 -> partition=5 offset=5
...
Done.

In the Confluent Cloud Console, open your topic and select the Messages tab. Filter by the recent timestamp range. Confirm 10 messages exist with the printed offsets and ISO 8601 timestamp keys.

Troubleshooting authentication failures

If the producer fails to authenticate, check the following:

Your IAM role has sts:GetWebIdentityToken and the role is attached to your AWS compute resource.
The audience value matches your Confluent Cloud OAuth identity provider configuration.
Your identity pool is mapped to the role-based access control (RBAC) role bindings required to produce to the topic.