Configure Azure User Assigned Managed Identity OAuth for Confluent Cloud

Confluent Cloud users can configure their client to use an Azure User-Assigned Managed Identity (UAMI) when connecting to an identity provider, typically Microsoft Entra ID, when using Apache Kafka® OAuth for authentication. Confluent Cloud UAMI support eliminates the need to manage static client IDs and secrets by leveraging Azure’s built-in identity management to automatically retrieve authentication tokens.

UAMIs are a feature of Microsoft Entra ID that eliminate the need to manage static credentials when connecting to Entra ID to receive the appropriate Confluent Cloud identity. UAMIs allow applications to obtain Microsoft Entra ID tokens for accessing resources that support Entra ID authentication, with Azure fully managing the identity lifecycle.

Benefits of using UAMI

Enhanced security

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

Simplified credential management

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

Seamless Azure integration

• Native integration with Azure resources and services
• Works with existing Azure RBAC and identity policies
• Supports cloud-native security best practices

Prerequisites

Before configuring UAMI for OAuth authentication, ensure you have:

• An Azure subscription with a User-Assigned Managed Identity and an Azure resource where your Kafka client will run, such as VM, AKS, or Azure Container Instance.
• Microsoft Entra ID configured with an App Registration. For details, see App registration in Microsoft Entra ID.
• Confluent Cloud only:
  • Access to Confluent Cloud with OrganizationAdmin role.
  • A Confluent Cloud Kafka cluster
• Confluent Platform only: Confluent Platform 8.1 or later.
• Java clients: version 8.1 or later of kafka-client-plugin.
• Non-Java clients: version 2.12 or higher of librdkafka.

Configure UAMI

Configure Microsoft Entra ID

Complete the following steps to configure UAMI in Microsoft Entra ID. For detailed Azure configuration steps, see the Microsoft Entra ID documentation.

1. Create an App Registration in Microsoft Entra ID. For instructions, see Register an application.
2. Enable V2 tokens by setting accessTokenAcceptedVersion to 2 in the App Registration manifest. For details, see Configure the application manifest.
3. Copy the Directory (tenant) ID from your App Registration Overview page. You need this to configure Confluent Cloud.

Create OAuth identity pools

1. Add Microsoft Entra ID as an OAuth/OIDC identity provider in Confluent Cloud. For complete instructions, see the documentation.

   When creating your identity provider:

   • Use the Directory (tenant) ID from your App Registration.
   • Click Import from Tenant ID to automatically populate the Issuer and JWKS URIs.

2. Create OAuth identity pools to map external identities to Confluent Cloud access policies. For instructions, see the documentation.

   For UAMI, include a filter to validate the token audience. In V2 tokens, the audience is the App Registration Application (client) ID:

   claims.aud == "<application_client_id>"
   

   Replace <application_client_id> with your App Registration’s Application (client) ID.

Complete Azure configuration for UAMI for Confluent Cloud

The following configuration steps are specific to UAMI. For detailed Azure configuration instructions, see the Azure IMDS documentation.

1. Set the Application ID URI in your App Registration to your Confluent Cloud cluster’s bootstrap endpoint, for example, pkc-abc123.us-west-2.aws.confluent.cloud:9092.

   For instructions, see Expose an API.

   Note

   In v2 tokens, the Application ID URI can be any valid URI. Using the bootstrap endpoint makes the configuration clearer and ensures proper token audience validation.

2. Optionally, configure app roles and API permissions in your App Registration as needed for your use case. For instructions, see Add app roles.

Verify configuration

Test your setup by retrieving an access token from your Azure resource.

For detailed instructions on testing UAMI token retrieval, see How to use managed identities.

After obtaining a token, verify that the iss field contains https://login.microsoftonline.com/[tenant_id]/v2.0. If it contains sts.windows.net, V1 tokens are in use and you need to update the App Registration manifest.

You can decode and inspect JWT tokens at jwt.io.

Configure Kafka clients

Configure Java clients

The UAMI functionality for Java clients is available in the kafka-client-plugins artifact. The earliest version that supports UAMI is version 8.1.

Add the dependency

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>latest</version>
    </dependency>
</dependencies>

Gradle:

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

dependencies {
    implementation 'io.confluent:kafka-client-plugins:latest'
}

Configure UAMI properties

Configure your Java client with the following UAMI-specific properties:

Properties props = new Properties();

// Basic OAuth configuration
props.put("bootstrap.servers", "<BOOTSTRAP_SERVERS>");
props.put("security.protocol", "SASL_SSL");
props.put("sasl.mechanism", "OAUTHBEARER");

// UAMI-specific configuration
props.put("sasl.oauthbearer.jwt.retriever.class",
    "io.confluent.security.auth.client.oauth.UamiJwtRetriever");
props.put("sasl.oauthbearer.token.endpoint.url",
    "http://169.254.169.254/metadata/identity/oauth2/token?api-version=<LATEST_VERSION>&resource=<BOOTSTRAP_ENDPOINT>&client_id=<UAMI_CLIENT_ID>");

// Set system property to allow the UAMI token endpoint
// The value must match the token endpoint URL including all parameters
System.setProperty("org.apache.kafka.sasl.oauthbearer.allowed.urls",
    "http://169.254.169.254/metadata/identity/oauth2/token?api-version=<LATEST_VERSION>&resource=<BOOTSTRAP_ENDPOINT>&client_id=<UAMI_CLIENT_ID>");

Replace the placeholders:

• <BOOTSTRAP_SERVERS>: Your Confluent Cloud bootstrap server URL.
• <LATEST_VERSION>: Azure IMDS API version. Use 2025-04-07 or later.
• <BOOTSTRAP_ENDPOINT>: Your Confluent Cloud cluster’s bootstrap server URL. This must match the Application ID URI you configured in the App Registration. You may need to URL-encode special characters such as colons (: becomes %3A) and forward slashes (/ becomes %2F).
• <UAMI_CLIENT_ID>: Your User-Assigned Managed Identity client ID.

Note

The client_id parameter in the token endpoint URL is required for Java clients. Without it, Azure will use the system-attached managed identity instead of the User-Assigned Managed Identity.

You can configure resource parameter as desired, but it is recommended that you configure to match your Application ID URI. If using the bootstrap endpoint as recommended, URL-encode special characters in the query string. For example, pkc-abc123.us-west-2.aws.confluent.cloud:9092 becomes pkc-abc123.us-west-2.aws.confluent.cloud%3A9092.

For more information about token URL parameters, see the Microsoft documentation on using VM tokens.

Java configuration parameters

Parameter	Description	Required
sasl.oauthbearer.jwt.retriever.class	JWT retriever class for UAMI. Must be set to io.confluent.security.auth.client.oauth.UamiJwtRetriever.	Yes
sasl.oauthbearer.token.endpoint.url	Azure IMDS token endpoint URL with query parameters including api-version, resource, and optionally client_id.	Yes
org.apache.kafka.sasl.oauthbearer.allowed.urls	System property: Allowlist for UAMI token endpoint. Must match the full token endpoint URL including parameters.	Yes

Configure non-Java (librdkafka-based) clients

For all Confluent Cloud-supported non-Java (librdkafka-based) clients such as Python, Go, .NET, JavaScript, and C/C++, configure your client with the following UAMI-specific properties:

Common configuration

Required properties:

• sasl.oauthbearer.metadata.authentication.type=azure_imds

  When configured, neither sasl.oauthbearer.client.id nor sasl.oauthbearer.client.secret are required.

• sasl.oauthbearer.config

  This is a general-purpose configuration property that accepts comma-separated key=value pairs. The query key is required, and its value is the GET query string to append to the token endpoint URL. The query string contains parameters required by Azure IMDS such as:

  • client_id: The UAMI client ID
  • resource: For determining the target audience
  • api-version: The API version to be used by the endpoint

Optional property:

• sasl.oauthbearer.token.endpoint.url

  This is set automatically when using sasl.oauthbearer.metadata.authentication.type=azure_imds but can be customized if needed.

Example configuration:

sasl.oauthbearer.metadata.authentication.type=azure_imds
sasl.oauthbearer.config=query=api-version=<LATEST_VERSION>&resource=<BOOTSTRAP_ENDPOINT>&client_id=<UAMI_CLIENT_ID>

For Confluent Cloud, you also need to add the logicalCluster and identityPoolId extensions:

sasl.oauthbearer.extensions=logicalCluster=<cluster-id>,identityPoolId=<pool-id>

Configure Python clients

Configure your Python client with UAMI-specific properties:

import logging
from confluent_kafka import Producer

def producer_config(args):
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)

    params = {
        'bootstrap.servers': args.bootstrap_servers,
        'security.protocol': 'SASL_SSL',
        'sasl.mechanisms': 'OAUTHBEARER',
        'sasl.oauthbearer.method': 'oidc',
        'sasl.oauthbearer.metadata.authentication.type': 'azure_imds',
        'sasl.oauthbearer.config': f'query={args.query}'
    }

    # These parameters are only applicable when producing to
    # Confluent Cloud where SASL extensions are required
    if args.logical_cluster and args.identity_pool_id:
        params['sasl.oauthbearer.extensions'] = (
            'logicalCluster=' + args.logical_cluster +
            ',identityPoolId=' + args.identity_pool_id
        )

    return params

# Create producer with UAMI configuration
producer_conf = producer_config(args)
producer = Producer(producer_conf)

Configure Go clients

Configure your Go client with UAMI-specific properties:

package main

import (
    "fmt"
    "github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
    bootstrapServers := "<BOOTSTRAP_SERVERS>"
    // Azure IMDS API version - use 2025-04-07 or later
    azureIMDSApiVersion := "2025-04-07"
    bootstrapEndpoint := "<BOOTSTRAP_ENDPOINT>"
    uamiClientId := "<UAMI_CLIENT_ID>"
    azureIMDSQueryParams := fmt.Sprintf("api-version=%s&resource=%s&client_id=%s",
        azureIMDSApiVersion, bootstrapEndpoint, uamiClientId)
    logicalCluster := "<your-logical-cluster>"
    identityPoolId := "<your-identity-pool-id>"

    config := fmt.Sprintf("query=%s", azureIMDSQueryParams)
    extensions := fmt.Sprintf("logicalCluster=%s,identityPoolId=%s",
        logicalCluster, identityPoolId)

    p, err := kafka.NewProducer(&kafka.ConfigMap{
        "bootstrap.servers":                              bootstrapServers,
        "security.protocol":                              "SASL_SSL",
        "sasl.mechanisms":                                "OAUTHBEARER",
        "sasl.oauthbearer.method":                        "oidc",
        "sasl.oauthbearer.metadata.authentication.type": "azure_imds",
        "sasl.oauthbearer.config":                        config,
        "sasl.oauthbearer.extensions":                    extensions,
    })

    if err != nil {
        panic(err)
    }
    defer p.Close()

    // Producer code here
}

Configure .NET clients

Configure your .NET client with UAMI-specific properties:

using Confluent.Kafka;
using System;
using System.Threading.Tasks;

public class UamiKafkaClient
{
    // Azure IMDS API version - use 2025-04-07 or later
    private const string azureIMDSApiVersion = "2025-04-07";
    private const string bootstrapEndpoint = "<BOOTSTRAP_ENDPOINT>";
    private const string uamiClientId = "<UAMI_CLIENT_ID>";
    private const string azureIMDSQueryParams =
        $"api-version={azureIMDSApiVersion}&resource={bootstrapEndpoint}&client_id={uamiClientId}";
    private const string kafkaLogicalCluster = "<your-logical-cluster>";
    private const string identityPoolId = "<your-identity-pool-id>";

    public static async Task Main(string[] args)
    {
        var bootstrapServers = args[0];
        var topicName = "test-topic";
        var groupId = Guid.NewGuid().ToString();

        var commonConfig = new ClientConfig
        {
            BootstrapServers = bootstrapServers,
            SecurityProtocol = SecurityProtocol.SaslSsl,
            SaslMechanism = SaslMechanism.OAuthBearer,
            SaslOauthbearerMethod = SaslOauthbearerMethod.Oidc,
            SaslOauthbearerMetadataAuthenticationType =
                SaslOauthbearerMetadataAuthenticationType.AzureIMDS,
            SaslOauthbearerConfig = $"query={azureIMDSQueryParams}",
            SaslOauthbearerExtensions =
                $"logicalCluster={kafkaLogicalCluster},identityPoolId={identityPoolId}"
        };

        // Use commonConfig with ProducerBuilder or ConsumerBuilder
        using (var producer = new ProducerBuilder<Null, string>(commonConfig).Build())
        {
            // Producer code here
        }
    }
}

Configure JavaScript clients

Configure your Node.js client with UAMI-specific properties using the confluent-kafka-javascript client:

const { Kafka } = require('@confluentinc/kafka-javascript').KafkaJS;

const bootstrapServers = '<BOOTSTRAP_SERVERS>';
// Azure IMDS API version - use 2025-04-07 or later
const azureIMDSApiVersion = '2025-04-07';
const bootstrapEndpoint = '<BOOTSTRAP_ENDPOINT>';
const uamiClientId = '<UAMI_CLIENT_ID>';
const azureIMDSQueryParams =
    `api-version=${azureIMDSApiVersion}&resource=${bootstrapEndpoint}&client_id=${uamiClientId}`;
const logicalCluster = '<your-logical-cluster>';
const identityPoolId = '<your-identity-pool-id>';

const kafka = new Kafka({
    'bootstrap.servers': bootstrapServers,
    'security.protocol': 'SASL_SSL',
    'sasl.mechanisms': 'OAUTHBEARER',
    'sasl.oauthbearer.method': 'oidc',
    'sasl.oauthbearer.metadata.authentication.type': 'azure_imds',
    'sasl.oauthbearer.config': `query=${azureIMDSQueryParams}`,
    'sasl.oauthbearer.extensions': `logicalCluster=${logicalCluster},identityPoolId=${identityPoolId}`
});

const producer = kafka.producer();
await producer.connect();

Note

The confluent-kafka-javascript client supports automatic UAMI token fetching through configuration properties, similar to Python, Go, and .NET clients. The client automatically retrieves tokens from the Azure Instance Metadata Service (IMDS) without requiring a custom callback function.

Non-Java (librdkafka-based) client configuration parameters

Parameter	Description	Required
sasl.oauthbearer.metadata.authentication.type	Set to azure_imds to enable UAMI authentication	Yes
sasl.oauthbearer.config	Query string parameters in format query=key1=value1&key2=value2. Must include api-version, resource, and client_id.	Yes
sasl.oauthbearer.extensions	Confluent Cloud extensions in format logicalCluster=<cluster-id>,identityPoolId=<pool-id>.	Yes, for Confluent Cloud
sasl.oauthbearer.token.endpoint.url	Azure IMDS endpoint URL. Defaults to http://169.254.169.254/metadata/identity/oauth2/token.	No

UAMI Examples

Python Example for UAMI

1. Configure your Python client with the following UAMI-specific properties

   def producer_config(args):
       logger = logging.getLogger(__name__)
       logger.setLevel(logging.DEBUG)
       params = {
           'bootstrap.servers': args.bootstrap_servers,
           'security.protocol': 'SASL_SSL',
           'sasl.mechanisms': 'OAUTHBEARER',
           'sasl.oauthbearer.method': 'oidc',
           'sasl.oauthbearer.metadata.authentication.type': 'azure_imds',
           'sasl.oauthbearer.config': f'query={args.query}'
       }
       # These two parameters are only applicable when producing to
       # confluent cloud where some sasl extensions are required.
       if args.logical_cluster and args.identity_pool_id:
           params['sasl.oauthbearer.extensions'] = 'logicalCluster=' + args.logical_cluster + \
               ',identityPoolId=' + args.identity_pool_id
   
       return params
   

2. Send these parameters into the producer’s constructor by calling the producer_config method from step 1.

   producer_conf = producer_config(args)
   producer = Producer(producer_conf)
   

.NET Example for UAMI

Configure your .NET client with the following UAMI-specific properties.

private const string azureIMDSQueryParams = "api-version=&resource=&client_id=";
        private const string kafkaLogicalCluster = "your-logical-cluster";
        private const string identityPoolId = "your-identity-pool-id";

        public static async Task Main(string[] args)
        {
            if (args.Length != 3)
            {
                Console.WriteLine("Usage: .. brokerList schemaRegistryUrl");
                return;
            }
            var bootstrapServers = args[1];
            var schemaRegistryUrl = args[2];
            var topicName = Guid.NewGuid().ToString();
            var groupId = Guid.NewGuid().ToString();

            var commonConfig = new ClientConfig
            {
                BootstrapServers = bootstrapServers,
                SecurityProtocol = SecurityProtocol.SaslPlaintext,
                SaslMechanism = SaslMechanism.OAuthBearer,
                SaslOauthbearerMethod = SaslOauthbearerMethod.Oidc,
                SaslOauthbearerMetadataAuthenticationType = SaslOauthbearerMetadataAuthenticationType.AzureIMDS,
                SaslOauthbearerConfig = $"query={azureIMDSQueryParams}",
                SaslOauthbearerExtensions = $"logicalCluster={kafkaLogicalCluster},identityPoolId={identityPoolId}"
            };

            var consumerConfig = new ConsumerConfig
            {
                BootstrapServers = bootstrapServers,
                SecurityProtocol = SecurityProtocol.SaslPlaintext,
                SaslMechanism = SaslMechanism.OAuthBearer,
                SaslOauthbearerMethod = SaslOauthbearerMethod.Oidc,
                GroupId = groupId,
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnableAutoOffsetStore = false
            };

            // pass the config values to the Producer's builder
            using (var producer = new ProducerBuilder<Null, User>(commonConfig)

JavaScript Example for UAMI

Configure your NodeJS client with the following UAMI-specific properties.

const bootstrapServers = "";
const azureIMDSQueryParams = "api-version=&resource=&client_id=";
const kafka = new Kafka({
    'bootstrap.servers': bootstrapServers,
    'security.protocol': 'SASL_SSL',
    'sasl.mechanisms': 'OAUTHBEARER',
    'sasl.oauthbearer.method': 'oidc',
    'sasl.oauthbearer.metadata.authentication.type': 'azure_imds',
    'sasl.oauthbearer.config': `query=${azureIMDSQueryParams}`
});

Go Example for UAMI

Configure your Go client with the following UAMI-specific properties

bootstrapServers := ""
azureIMDSQueryParams := "api-version=&resource=&client_id=";
config := fmt.Sprintf("query=%s", azureIMDSQueryParams)
p, err := kafka.NewProducer(&kafka.ConfigMap {
    "bootstrap.servers": bootstrapServers,
    "security.protocol": "SASL_SSL",
    "sasl.mechanisms": "OAUTHBEARER",
    "sasl.oauthbearer.method": "oidc",
    "sasl.oauthbearer.metadata.authentication.type": "azure_imds",
    "sasl.oauthbearer.config": config
})

Best practices

Security

• Use separate User-Assigned Managed Identities for different applications or environments.
• Follow the principle of least privilege when assigning Azure roles and permissions.
• Regularly audit managed identity usage and access patterns.
• Enable Microsoft Entra ID logging and monitoring for identity-related events.
• Rotate App Registration secrets regularly if used for testing or fallback scenarios.

Operations

• Use descriptive names for managed identities to improve manageability.
• Document the purpose and scope of each managed identity.
• Implement proper error handling and retry logic in your applications.
• Test failover scenarios and token refresh behavior.
• Monitor authentication success rates and token retrieval latency.

Architecture

• Design identity pools with clear boundaries and purposes.
• Use identity pool filters to automatically map identities based on JWT claims.
• Consider using multiple identity pools for different access levels or teams.
• Implement monitoring and alerting for authentication failures.
• Plan for identity lifecycle management as applications are deployed and decommissioned.

For comprehensive OAuth best practices, see the documentation.

Limitations

The UAMI OAuth feature has the following limitations:

• Azure-only: Available only for applications running on Azure resources with managed identity support.
• Network connectivity: Requires network access to Azure Instance Metadata Service (IMDS) at 169.254.169.254.
• Identity provider: Works only with Microsoft Entra ID as the identity provider.
• Client versions:
  • Java: Requires kafka-client-plugins version 8.1.0 or later.
  • Non-Java (librdkafka-based): Requires librdkafka version 2.12.0 or later.
• Azure quotas: Subject to Azure Managed Identity service limits and quotas.

Related content

• What is managed identities for Azure resources?
• Managed identities for Azure resources documentation
• Azure Instance Metadata Service
• Microsoft Entra ID documentation
• Microsoft identity platform documentation
• How to use managed identities for Azure resources on an Azure VM to acquire an access token
• Use OAuth/OIDC to Authenticate to Confluent Cloud
• Use OAuth Identity Pools with Your OAuth/OIDC Identity Provider on Confluent Cloud
• Add an identity provider using Confluent Cloud Console