<a id="oauth-uami-share"></a>

# Configure Azure User Assigned Managed Identity OAuth for Confluent Platform

Confluent Platform 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 Platform 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 Platform 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](https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app).
- Confluent Cloud only:
  - Access to Confluent Cloud with [OrganizationAdmin](https://docs.confluent.io/cloud/current/security/access-control/rbac/predefined-rbac-roles.html#organizationadmin-role)
    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](https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/overview).

1. Create an App Registration in Microsoft Entra ID. For instructions, see
   [Register an application](https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app).
2. Enable V2 tokens by setting `accessTokenAcceptedVersion` to `2` in
   the App Registration manifest. For details, see
   [Configure the application manifest](https://learn.microsoft.com/en-us/entra/identity-platform/reference-app-manifest#configure-the-app-manifest).
3. Copy the **Directory (tenant) ID** from your App Registration Overview
   page. You need this ID for later configuration steps.

### Configure Confluent Platform brokers

Configure your Confluent Platform brokers to trust Microsoft EntraID tokens. For complete
broker configuration details, see [Configure Confluent Server Brokers for OAuth Authentication in Confluent Platform](../../oauth-oidc/configure-cs.md#configure-cs-for-oauth).

### Complete Azure configuration for UAMI for Confluent Platform

The following configuration steps are specific to UAMI. For detailed Azure
configuration instructions, see the
[Azure IMDS documentation](https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/how-to-use-vm-token).

1. Set the **Application ID URI** in your App Registration to your
   Apache Kafka® cluster’s bootstrap endpoint.

   For example, `kafka.example.com:9092`.

   For instructions, see
   [Expose an API](https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-configure-app-expose-web-apis#add-a-scope).

   #### 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](https://learn.microsoft.com/en-us/entra/identity-platform/howto-add-app-roles-in-apps).

### 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](https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/how-to-use-vm-token).

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](https://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:**

```xml
<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:**

```groovy
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:

```java
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 Kafka cluster’s bootstrap server URL.
- `<latest_version>`: Azure IMDS API version. Use `2025-04-07` or
  later.
- `<bootstrap_endpoint>`: Your Kafka 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, `kafka.example.com:9092` becomes
`kafka.example.com%3A9092`.

For more information about token URL parameters, see the
[Microsoft documentation on using VM tokens](https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/how-to-use-vm-token#get-a-token-using-http).

## Java configuration parameters

| Parameter                                        | Description                                                                                                              | Required   |
|--------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------|------------|
| `sasl.oauthbearer.jwt.retriever.class`           | JWT retriever class for UAMI. Must be set to<br/>`io.confluent.security.auth.client.oauth.UamiJwtRetriever`.             | Yes        |
| `sasl.oauthbearer.token.endpoint.url`            | Azure IMDS token endpoint URL with query parameters including<br/>`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<br/>full token endpoint URL including parameters.     | Yes        |

## Configure non-Java (librdkafka-based) clients

For all 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:**

```properties
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:

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

### Configure Python clients

Configure your Python client with UAMI-specific properties:

```python
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:

```go
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:

```csharp
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:

```javascript
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`.<br/>Must include `api-version`, `resource`, and `client_id`. | Yes                      |
| `sasl.oauthbearer.extensions`                   | Confluent Cloud extensions in format<br/>`logicalCluster=<cluster-id>,identityPoolId=<pool-id>`.                                | Yes, for Confluent Cloud |
| `sasl.oauthbearer.token.endpoint.url`           | Azure IMDS endpoint URL. Defaults to<br/>`http://169.254.169.254/metadata/identity/oauth2/token`.                               | No                       |

## UAMI examples

### Python example

1. Configure your Python client with the following UAMI-specific properties
   ```python
   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.
   ```python
   producer_conf = producer_config(args)
   producer = Producer(producer_conf)
   ```

### .NET example

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

```c#
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

Configure your Node.js client with the following UAMI-specific properties.

```javascript
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

Configure your Go client with the following UAMI-specific properties

```go
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](https://docs.confluent.io/cloud/current/security/authenticate/workload-identities/identity-providers/oauth/best-practices.html#best-practices-oauth-identity-providers).

## 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?](https://learn.microsoft.com/en-us/azure/active-directory/managed-identities-azure-resources/overview)
- [Managed identities for Azure resources documentation](https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/)
- [Azure Instance Metadata Service](https://learn.microsoft.com/en-us/azure/virtual-machines/instance-metadata-service)
- [Microsoft Entra ID documentation](https://learn.microsoft.com/en-us/entra/identity/)
- [Microsoft identity platform documentation](https://learn.microsoft.com/en-us/azure/active-directory/develop/)
- [How to use managed identities for Azure resources on an Azure VM to acquire an access token](https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/how-to-use-vm-token)

- [Configure Confluent Server Brokers for OAuth Authentication in Confluent Platform](../../oauth-oidc/configure-cs.md#configure-cs-for-oauth)
- [Configure Metadata Service (MDS) for OAuth Authentication in Confluent Platform](../../oauth-oidc/configure-mds.md#configure-mds-for-oauth)
- [Use Role-Based Access Control (RBAC) for Authorization in Confluent Platform](../../../authorization/rbac/overview.md#rbac-overview)
- [Use OAuth/OIDC for Authentication in Confluent Platform](../../oauth-oidc/overview.md#oauth-oidc-authentication-overview)