Configure AWS IAM OAuth for Java Client

Confluent Platform 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 Platform 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 Platform authorization. Authorization is handled by Confluent Platform 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 Platform 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 Platform OAuth identity provider is configured to trust your AWS account’s issuer URL and reference AWS’s JWKS URI.

• A Confluent Platform identity pool is configured and mapped to the Confluent Platform RBAC role bindings your client needs.

• Java Client: version 8.3 or later of kafka-client-plugins.

• Confluent Platform 8.3 or later.

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 Platform cluster bootstrap URL, including port. Must match the audience configured on your Confluent Platform 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 Platform Kafka cluster ID, for example lkc-xxxxxx.
JAAS extension_identityPoolId	Yes	The Confluent Platform identity pool ID this client authenticates through, for example pool-xxxxxx.

Important

The audience value must exactly match what your Confluent Platform OAuth identity provider is configured to accept. If authentication fails with an audience-mismatch error, check the identity provider configuration in the Confluent Platform 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.

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

2. 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.");
       }
   }
   

3. 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.
   

4. In the Confluent Platform 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 Platform OAuth identity provider configuration.

• Your identity pool is mapped to the role-based access control (RBAC) role bindings required to produce to the topic.

Related content

• Configure Azure User Assigned Managed Identity OAuth for Confluent Platform

• kafka-client-plugins Maven repository

• AWS documentation: OIDC federation

• AWS blog: Simplify access to external services using AWS IAM Outbound Identity Federation

• AWS documentation: GetWebIdentityToken API reference

• AWS documentation: GetWebIdentityTokenRequest Java API reference