Skip to content

MuleSoft Producer & Consumer Connector and Confluent Schema Registry Setup with Kafka OAuth

This guide explains how to configure Mule Kafka producers and consumers that authenticate to Confluent Cloud with OAuth (OAUTHBEARER) and publish/consume messages validated against the Confluent Schema Registry (also using OAuth).

Replace every <placeholder> below with values from your own Confluent Cloud environment and identity provider.

Prerequisites

Before proceeding, ensure you have completed the following:

  1. Azure AD App Registration - Follow the OAuth Client Identity Pools guide to create an enterprise app registration
  2. API Permissions - Ensure your app registration has ServicePrincipal.Authenticate permission on the kaas-oauth resource
  3. MuleSoft Runtime - MuleSoft 4.x runtime with Kafka connector support
  4. Kafka Bootstrap URL - The cluster bootstrap URL from Confluent Cloud UI

OAuth attribute rules (applies to producer and consumer)

Value Where to find it Example Format Notes
Bootstrap server Cluster → Cluster settings pkc-zd997.eu-central-1.aws.confluent.cloud:9092 MUST include :9092 port — this is critical!
Schema Registry URL Environment → Schema Registry → API endpoint https://psrc-xxxxx.<region>.<cloud>.confluent.cloud Uses HTTPS protocol
SR logical cluster ID Environment → Schema Registry → Settings lsrc-xxxxx From Schema Registry settings
OAuth client id / secret Your identity provider (e.g. Azure AD app registration) <oauth-client-id> / <oauth-client-secret> Keep secrets secure in ${secure:...}
OAuth token endpoint Identity provider (Azure Entra ID) https://login.microsoftonline.com/<tenant>/oauth2/v2.0/token Microsoft Entra endpoint
OAuth scope Identity provider / API exposed scope api://af713fe2-aab8-4437-8db5-47df799d134b/.default KaaS OAuth resource app scope
Broker audience Confluent OAuth/OIDC integration pkc-zd997.eu-central-1.aws.confluent.cloud:9092 Same as bootstrap server

Producer configuration

Add a kafka:producer-config that uses producer-sasl-oauth-bearer-cc-connection. The broker OAuth attributes go on the connection element; the schema registry OAuth properties go inside <kafka:additional-properties>.

Visual Configuration Reference (MuleSoft Studio):

Apache Kafka Producer OAuth Configuration - General Tab

Apache Kafka Producer OAuth Configuration - Advanced Tab with Schema Registry

XML Configuration:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
<kafka:producer-config name="Kafka_Producer_OAuth"
        doc:name="Apache Kafka Producer configuration">
    <kafka:producer-sasl-oauth-bearer-cc-connection
            clientId="${kafka.oauth.clientId}"
            clientSecret="${kafka.oauth.clientSecret}"
            oauthTokenEndpoint="${kafka.oauth.tokenEndpointUrl}"
            tlsContext="TLS_Context_Kafka"
            scope="${kafka.oauth.scope}"
            audience="${kafka.bootstrap.server}">
        <kafka:bootstrap-servers>
            <kafka:bootstrap-server value="${kafka.bootstrap.server}" />
        </kafka:bootstrap-servers>

        <!-- Schema Registry (OAuth) + Avro serializer -->
        <kafka:additional-properties>
            <kafka:additional-property key="schema.registry.url"
                value="${kafka.schema.url}" />
            <kafka:additional-property key="bearer.auth.credentials.source"
                value="OAUTHBEARER" />
            <kafka:additional-property key="bearer.auth.issuer.endpoint.url"
                value="${kafka.oauth.tokenEndpointUrl}" />
            <kafka:additional-property key="bearer.auth.client.id"
                value="${kafka.oauth.clientId}" />
            <kafka:additional-property key="bearer.auth.client.secret"
                value="${kafka.oauth.clientSecret}" />
            <kafka:additional-property key="bearer.auth.scope"
                value="${kafka.oauth.scope}" />
            <kafka:additional-property key="bearer.auth.logical.cluster"
                value="${kafka.schema.logicalCluster}" />
            <kafka:additional-property key="value.subject.name.strategy"
                value="io.confluent.kafka.serializers.subject.TopicNameStrategy" />
            <kafka:additional-property key="value.serializer"
                value="com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer" />
            <kafka:additional-property key="auto.register.schemas"
                value="true" />
        </kafka:additional-properties>
    </kafka:producer-sasl-oauth-bearer-cc-connection>
</kafka:producer-config>

Note: The schema registry configuration shown above uses Avro serialization. You can use the same OAuth authentication approach for JSON and Protobuf formats by replacing the serializer and related properties.

The OAuth bearer authentication properties (bearer.auth.* and schema.registry.url) remain the same across all formats.

Consumer configuration (OAuth)

A consumer uses the matching kafka:consumer-sasl-oauth-bearer-cc-connection. The OAuth attributes work the same way as the producer.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
<kafka:consumer-config name="Kafka_Consumer_OAuth"
        doc:name="Apache Kafka Consumer configuration">
    <kafka:consumer-sasl-oauth-bearer-cc-connection
            groupId="${kafka.consumer.groupId}"
            clientId="${kafka.oauth.clientId}"
            clientSecret="${kafka.oauth.clientSecret}"
            oauthTokenEndpoint="${kafka.oauth.tokenEndpointUrl}"
            scope="${kafka.oauth.scope}"
            audience="${kafka.bootstrap.server}"
            tlsContext="TLS_Context_Kafka">
        <kafka:bootstrap-servers>
            <kafka:bootstrap-server value="${kafka.bootstrap.server}" />
        </kafka:bootstrap-servers>
        <kafka:topic-patterns>
            <kafka:topic-pattern value="${kafka.topic}" />
        </kafka:topic-patterns>
    </kafka:consumer-sasl-oauth-bearer-cc-connection>
</kafka:consumer-config>

Consuming Avro (optional)

To deserialize Avro from the Schema Registry, add <kafka:additional-properties> with the registry OAuth settings and the Avro deserializer:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
<kafka:additional-properties>
    <kafka:additional-property key="schema.registry.url"            value="${kafka.schema.url}" />
    <kafka:additional-property key="bearer.auth.credentials.source" value="OAUTHBEARER" />
    <kafka:additional-property key="bearer.auth.issuer.endpoint.url" value="${kafka.oauth.tokenEndpointUrl}" />
    <kafka:additional-property key="bearer.auth.client.id"          value="${kafka.oauth.clientId}" />
    <kafka:additional-property key="bearer.auth.client.secret"      value="${kafka.oauth.clientSecret}" />
    <kafka:additional-property key="bearer.auth.scope"              value="${kafka.oauth.scope}" />
    <kafka:additional-property key="bearer.auth.logical.cluster"    value="${kafka.schema.logicalCluster}" />
    <kafka:additional-property key="value.deserializer"             value="io.confluent.kafka.serializers.KafkaAvroDeserializer" />
</kafka:additional-properties>

Note: The schema registry configuration shown above uses Avro deserialization. You can use the same OAuth authentication approach for JSON and Protobuf formats by replacing the deserializer and related properties

The OAuth bearer authentication properties (bearer.auth.* and schema.registry.url) remain the same across all formats.

Consumer Configuration Reference

The screenshots below show how these properties map to the MuleSoft Studio visual configuration for consumers:

Apache Kafka Consumer OAuth Configuration - General Tab MuleSoft Studio - Kafka Consumer Configuration with OAuth settings

Apache Kafka Consumer OAuth Configuration - Advanced Tab with Schema Registry MuleSoft Studio - Advanced Configuration with Schema Registry OAuth properties

Property reference

Property Purpose
schema.registry.url Confluent Cloud Schema Registry endpoint
bearer.auth.credentials.source=OAUTHBEARER Tells the SR client to use OAuth
bearer.auth.issuer.endpoint.url OAuth token endpoint used by the SR client
bearer.auth.client.id / client.secret OAuth credentials for SR
bearer.auth.scope OAuth scope for SR
bearer.auth.logical.cluster SR logical cluster id (lsrc-...) — required for Confluent Cloud
value.subject.name.strategy TopicNameStrategy → subject = <topic>-value
value.serializer MuleKafkaAvroSerializer — Mule's Avro serializer wrapper
auto.register.schemas true registers schema on first publish; false requires it to be pre-registered

Security: move clientSecret and bearer.auth.client.secret to property files (e.g. ${kafka.oauth.clientSecret}) or the Secure Properties module. Do not commit real secrets to source control.

Using Postman or cURL

Test the OAuth token endpoint directly:

1
2
3
4
5
6
curl -X POST "https://login.microsoftonline.com/<tenant>/oauth2/v2.0/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "client_id=<your-client-id>" \
  -d "client_secret=<your-client-secret>" \
  -d "scope=api://af713fe2-aab8-4437-8db5-47df799d134b/.default" \
  -d "grant_type=client_credentials"

Resources