Explicit Topic-Subject Associations

An explicit topic-subject association links a Kafka topic directly to a Schema Registry subject.

Note

Explicit topic-subject associations are in Limited Availability and are available on Dedicated clusters. To enable them for your environment, contact your Confluent account team.

Clients use a subject name strategy to decide which subject to register a topic’s schema under and read it from:

Strategy	Association type	How it finds the subject	Subject name	When it applies
`AssociatedNameStrategy`	Explicit	Looks up the subject associated with the topic	The subject the association points to	The default on client library versions that support it. If the topic has no association, it falls back to `TopicNameStrategy`.
`TopicNameStrategy`	Legacy implicit	Derives the name from the topic	`<topic_name>-key` or `<topic_name>-value`	The default on earlier client library versions, and the default fallback. All messages in a topic use the same schema.
`RecordNameStrategy`	Legacy implicit	Derives the name from the record	The fully qualified record name	A topic carries more than one record type, and you want compatibility checked per record type across all topics.
`TopicRecordNameStrategy`	Legacy implicit	Derives the name from the topic and the record	`<topic_name>-<fully_qualified_record_name>`	A topic carries more than one record type, and you want compatibility checked per record type within each topic.

The strategies other than AssociatedNameStrategy derive the subject name from the topic or the record each time a client runs. Nothing records which subject a topic uses, and nothing stops someone from deleting that subject while the topic still uses it. An association records the relationship and prevents deletion of the subject while a topic uses it.

For the client library versions that support AssociatedNameStrategy, see Supported client library versions. For details on the other legacy strategies, see Subject name strategy.

Explicit associations introduce these distinct architectural patterns:

Topic-native schemas, also called native schemas, where a schema is strictly bound to a single topic. The schema exists only while the topic exists, like a table schema in a database.
Shared schemas, where a single schema can be explicitly linked across more than one topic. Schema Registry subjects already work this way. The subject stays when you remove the association.

A topic created with a schema gets a native schema, and a topic associated with an existing subject gets a shared schema. The API reports each association’s lifecycle, STRONG for native and WEAK for shared. See Key concepts.

Key concepts

Term	Meaning
Subject	The named container in Schema Registry where a schema and its versions are registered. For example, `orders-value`.
Association	The link between a topic and a subject. It has a lifecycle of `STRONG` or `WEAK`.
Native schema	A schema owned by one topic. It has a lifecycle of `STRONG`.
Shared schema	A schema that more than one topic can use. It has a lifecycle of `WEAK`.
`AssociatedNameStrategy`	The subject name strategy that clients use by default on supported client library versions. It uses a topic’s association to find the subject, and falls back to `TopicNameStrategy` if the topic doesn’t have an association.

How native and shared schemas differ

	Native schema (`STRONG`)	Shared schema (`WEAK`)
How you create it	Together with the topic, through the Kafka API.	Register the subject in Schema Registry, then associate a topic with it.
Topics that can use it	One	Many
Subject name	By default, `<topic_name>-value`, in a schema context named after the cluster ID.	Any existing subject, whatever its name.
Reserves the subject	Yes. Schema Registry rejects the association if the subject already has any other association, and rejects any other association on that subject afterward. It checks by subject name across all clusters in the environment.	No
Subject deletion	Blocked while the topic exists.	Blocked while any topic has an association.
When you delete the topic	Schema Registry deletes the association, the subject, and every schema version.	Schema Registry removes the association. The subject and its schema stay.
Removing the association	Only by deleting the topic.	Unlink the topic at any time.

You can associate an existing subject only with a topic that doesn’t already have a native schema.

If two topics with the same name exist in different clusters of the same environment, such as a source topic and its Cluster Linking mirror, they share the same subject name, <topic_name>-value. Only one of them can hold a STRONG association on it. Associate a different subject for the other.

Prerequisites

A Confluent Cloud environment with a Dedicated Kafka cluster and Schema Registry enabled. Your Confluent account team enables explicit topic-subject associations for your environment.
Client libraries that support AssociatedNameStrategy. See Supported client library versions.
A Kafka API key and a Schema Registry API key. Kafka and Schema Registry keep separate endpoints, credentials, and authorization. See Authorization.

The following examples use three resources:

orders: a topic with its own native Order schema.
Customer: a shared schema registered on its own.
customer-signups: a topic associated with the shared Customer schema.

Create a topic with a native schema

Create the topic and its schema in one request through the Kafka API. The schema belongs to the topic. It evolves with the topic, and Schema Registry deletes it when you delete the topic.

By default, Schema Registry names the subject <topic_name>-value and places it in a schema context named after the cluster ID. For example, :.lkc-abc123:orders-value.

Note

Deleting a topic that has a native association also deletes its subject and every schema version.

These examples use the Order schema:

{
  "type": "record",
  "name": "Order",
  "namespace": "com.example.avro",
  "fields": [
    { "name": "orderId", "type": "string" },
    { "name": "amount", "type": "double" }
  ]
}

Cloud Console

Go to Environment > Cluster > Topics > Add topic.
On Set topic settings, name the topic orders, set the number of partitions, and click Next.

On Associate schema (Optional), under Create a schema (message value), the Subject name is pre-filled with the generated, context-scoped name and can’t be edited.

To skip schema association and create the topic with no schema, click Create without schema.
Set Type to Avro and paste the Order schema. Click Validate.
Click Next, then finish the wizard.

The schema appears on the topic’s Schema tab. It also appears under Schemas in Schema Registry, scoped to the cluster’s schema context.

Confluent CLI

Run this command to create the orders topic with the Order schema as its native value schema:

confluent kafka topic create orders --partitions 1 \
  --config 'confluent.value.association={"schemaType":"AVRO",'\
'"schema":"{\"type\":\"record\",\"name\":\"Order\",'\
'\"namespace\":\"com.example.avro\",'\
'\"fields\":[{\"name\":\"orderId\",\"type\":\"string\"},'\
'{\"name\":\"amount\",\"type\":\"double\"}]}"}'

To create a key schema, use confluent.key.association.

REST API

Send this request to create the orders topic with the Order schema as its native value schema:

POST /kafka/v3/clusters/{cluster_id}/topics
Content-Type: application/json

{
  "topic_name": "orders",
  "partitions_count": 1,
  "configs": [
    {
      "name": "confluent.value.association",
      "value": "{\"schemaType\":\"AVRO\",\"schema\":\"{\\\"type\\\":\\\"record\\\",\\\"name\\\":\\\"Order\\\",\\\"namespace\\\":\\\"com.example.avro\\\",\\\"fields\\\":[{\\\"name\\\":\\\"orderId\\\",\\\"type\\\":\\\"string\\\"},{\\\"name\\\":\\\"amount\\\",\\\"type\\\":\\\"double\\\"}]}\"}"
    }
  ]
}

Terraform

Define the orders topic with the Order schema in the config map:

resource "confluent_kafka_topic" "orders" {
  kafka_cluster {
    id = confluent_kafka_cluster.main.id
  }
  topic_name       = "orders"
  partitions_count = 1
  rest_endpoint    = confluent_kafka_cluster.main.rest_endpoint

  config = {
    "confluent.value.association" = jsonencode({
      schemaType = "AVRO"
      schema = jsonencode({
        type      = "record"
        name      = "Order"
        namespace = "com.example.avro"
        fields = [
          { name = "orderId", type = "string" },
          { name = "amount", type = "double" }
        ]
      })
    })
  }

  credentials {
    key    = confluent_api_key.env-manager.id
    secret = confluent_api_key.env-manager.secret
  }
}

Java (AdminClient)

Create the orders topic with the Order schema through the AdminClient:

Properties props = new Properties();
props.put("bootstrap.servers", "<kafka-endpoint>");

try (AdminClient adminClient = AdminClient.create(props)) {
    NewTopic topic = new NewTopic("orders", 1, (short) 1);

    Map<String, String> valueAssociation = new HashMap<>();
    valueAssociation.put("schemaType", "AVRO");
    valueAssociation.put("schema",
        "{\"type\":\"record\",\"name\":\"Order\","
        + "\"namespace\":\"com.example.avro\","
        + "\"fields\":[{\"name\":\"orderId\",\"type\":\"string\"},"
        + "{\"name\":\"amount\",\"type\":\"double\"}]}");

    Map<String, String> topicConfig = new HashMap<>();
    topicConfig.put("confluent.value.association",
        new ObjectMapper().writeValueAsString(valueAssociation));
    topic.configs(topicConfig);

    adminClient.createTopics(Collections.singletonList(topic)).all().get();
}

Use a shared schema across topics

A shared schema is a subject that you register one time and associate with any topic that doesn’t already have a native schema. Other topics can keep using it, and Schema Registry blocks deletion of the subject while any topic has an association with it.

To share a schema, register the shared schema, and then associate a topic with it.

Register the shared schema

Registering a shared schema is ordinary Schema Registry registration. To learn how to register a schema in the Cloud Console, see Create a schema. These examples use the Customer schema:

{
  "type": "record",
  "name": "Customer",
  "namespace": "com.example.avro",
  "fields": [
    { "name": "customerId", "type": "string" },
    { "name": "email", "type": "string" }
  ]
}

Cloud Console

Go to Environment > Schema Registry > Add schema.
Set the subject to Customer, set the format to Avro, and paste the schema.
Click Create.

Confluent CLI

Save the Customer schema as customer.avsc, and then run this command to register it:

confluent schema-registry schema create \
  --subject Customer \
  --schema customer.avsc \
  --type avro

REST API

Send this request to register the Customer schema:

POST /subjects/Customer/versions
Content-Type: application/vnd.schemaregistry.v1+json

{
  "schemaType": "AVRO",
  "schema": "{\"type\":\"record\",\"name\":\"Customer\",\"namespace\":\"com.example.avro\",\"fields\":[{\"name\":\"customerId\",\"type\":\"string\"},{\"name\":\"email\",\"type\":\"string\"}]}"
}

Terraform

Define the Customer schema as a confluent_schema resource:

resource "confluent_schema" "customer" {
  schema_registry_cluster {
    id = confluent_schema_registry_cluster.main.id
  }
  rest_endpoint = confluent_schema_registry_cluster.main.rest_endpoint
  subject_name  = "Customer"
  format        = "AVRO"
  schema        = file("${path.module}/customer.avsc")

  credentials {
    key    = confluent_api_key.sr-manager.id
    secret = confluent_api_key.sr-manager.secret
  }
}

You can also register shared schemas from a CI/CD pipeline with the Schema Registry Maven or Gradle plugin.

Associate a topic with the shared schema

If you give the topic a subject and no schema definition, Confluent Cloud associates an existing subject and doesn’t create a native schema. The lifecycle is always WEAK.

Cloud Console

Use any of these approaches to associate a topic with the shared schema:

When you create the topic:
1. Go to Environment > Cluster > Topics > Add topic.
2. Name the topic, and click Next.
3. On Associate schema (Optional), click Associate existing schema.
4. Choose the Customer subject, and click Next.
On an existing topic:
1. Go to Topics > customer-signups > Schema tab > Value.
2. Click Associate schema.
3. Under Subject name, search for Customer and select it.
4. Click Save.
From the subject:
1. Go to Schemas > Customer > Associate > Associate existing topic.
2. Choose customer-signups.
3. Click Associate.

Confluent CLI

Run this command to associate the customer-signups topic with the Customer subject:

confluent kafka topic update customer-signups \
  --config 'confluent.value.association={"subject":"Customer"}'

REST API

Send this request to associate the customer-signups topic with the Customer subject:

POST /kafka/v3/clusters/{cluster_id}/topics/customer-signups/configs:alter
Content-Type: application/json

{
  "data": [
    {
      "name": "confluent.value.association",
      "operation": "SET",
      "value": "{\"subject\":\"Customer\"}"
    }
  ]
}

Terraform

Define the customer-signups topic with an association to the Customer subject:

resource "confluent_kafka_topic" "customer_signups" {
  kafka_cluster {
    id = confluent_kafka_cluster.main.id
  }
  topic_name       = "customer-signups"
  partitions_count = 1
  rest_endpoint    = confluent_kafka_cluster.main.rest_endpoint

  config = {
    "confluent.value.association" = jsonencode({
      subject = "Customer"
    })
  }

  credentials {
    key    = confluent_api_key.env-manager.id
    secret = confluent_api_key.env-manager.secret
  }
}

Java (AdminClient)

Associate the customer-signups topic with the Customer subject through the AdminClient:

Map<String, String> valueAssociation = new HashMap<>();
valueAssociation.put("subject", "Customer");

ConfigResource topic =
    new ConfigResource(ConfigResource.Type.TOPIC, "customer-signups");
AlterConfigOp op = new AlterConfigOp(
    new ConfigEntry("confluent.value.association",
        new ObjectMapper().writeValueAsString(valueAssociation)),
    AlterConfigOp.OpType.SET);

adminClient.incrementalAlterConfigs(Map.of(topic, List.of(op))).all().get();

Remove an association

To remove a shared association, unlink the topic. The shared subject and its schema stay in Schema Registry, and other topics keep using them. After you remove the association, the topic’s confluent.value.association configuration returns to its default empty value. Schema Registry no longer lists the association for the topic.

You can’t remove a native association on its own. Schema Registry removes it when you delete the topic, which also deletes the subject.

To remove a key association, use confluent.key.association for the configuration name.

Confluent CLI

Run this command to remove the association from the customer-signups topic:

confluent kafka topic update customer-signups \
  --config 'confluent.value.association=null'

REST API

Send this request to remove the association from the customer-signups topic:

POST /kafka/v3/clusters/{cluster_id}/topics/customer-signups/configs:alter
Content-Type: application/json

{
  "data": [
    { "name": "confluent.value.association", "operation": "DELETE" }
  ]
}

A successful request returns 204 No Content.

Terraform

Remove only the confluent.value.association entry from the topic’s config map, and keep the other entries. For example, a topic that also sets retention.ms changes from this:

config = {
  "retention.ms"                = "604800000"
  "confluent.value.association" = jsonencode({ subject = "Customer" })
}

to this:

config = {
  "retention.ms" = "604800000"
}

Then apply the change.

Java (AdminClient)

Remove the association from the customer-signups topic through the AdminClient:

ConfigResource topic =
    new ConfigResource(ConfigResource.Type.TOPIC, "customer-signups");
AlterConfigOp op = new AlterConfigOp(
    new ConfigEntry("confluent.value.association", null),
    AlterConfigOp.OpType.DELETE);

adminClient.incrementalAlterConfigs(Map.of(topic, List.of(op))).all().get();

Query associations

Use the Schema Registry Associations API to find the subjects associated with a topic or the topics associated with a subject. Send each request to your Schema Registry endpoint, authenticated with a Schema Registry API key.

Look up by subject

Send this request to list the topics associated with the Customer subject:

GET /associations/subjects/Customer

Example response:

[
  {
    "subject": "Customer",
    "guid": "7a3c1f52-2d4e-4b8a-b6c1-9e0f5a1d2c33",
    "resourceName": "customer-signups",
    "resourceNamespace": "lkc-abc123",
    "resourceId": "ZyXwVuTsRqPoNmLkJiHgFe",
    "resourceType": "topic",
    "associationType": "value",
    "lifecycle": "WEAK",
    "createTs": 1719000000000,
    "updateTs": 1719000000000
  }
]

Look up by topic

Send this request to list the subjects associated with the orders topic:

GET /associations/resources/{cluster_id}/orders

{cluster_id} is the Kafka cluster ID (lkc-...).

Example response:

[
  {
    "subject": ":.lkc-abc123:orders-value",
    "guid": "5f4d6d3e-8b1a-4c2e-9a7f-1c9e8d2b3a4f",
    "resourceName": "orders",
    "resourceNamespace": "lkc-abc123",
    "resourceId": "AbCdEfGhIjKlMnOpQrStUv",
    "resourceType": "topic",
    "associationType": "value",
    "lifecycle": "STRONG",
    "createTs": 1719000000000,
    "updateTs": 1719000000000
  }
]

createTs and updateTs are epoch milliseconds. guid is an internal identifier, not a schema ID.

Filters and paging

Both lookups accept these optional query parameters:

Parameter	Values	Example
`resourceType`	`topic`	`GET /associations/subjects/Customer?resourceType=topic`
`associationType`	`key` or `value`	`GET /associations/subjects/Customer?associationType=key`
`lifecycle`	`STRONG` or `WEAK`	`GET /associations/subjects/Customer?lifecycle=WEAK`
`offset`	Where to start. Defaults to `0`.	`GET /associations/subjects/Customer?offset=1`
`limit`	The most results to return. Defaults to `-1`. Any negative value returns all results.	`GET /associations/subjects/Customer?limit=1`

An empty list means that no association matches your filters or that your principal can’t read any of the matching subjects. A change to an association can take a few seconds to appear in lookup results. To look up by subject, you need read access to the subject. To look up by topic, you need to be an authenticated principal. Schema Registry returns the associations whose subjects you can read.

Delete topics and subjects

Action	Native schema (`STRONG`)	Shared schema (`WEAK`)
Delete the topic	Deletes the association, the subject, and all schema versions.	Removes the association. The subject and its schema stay.
Delete the subject	Blocked while the topic exists.	Blocked while any topic has an association. Unlink the topics first.

Use associations with clients

On a supported client library version, AssociatedNameStrategy is the default subject name strategy, and you don’t need to configure anything. The strategy finds the topic’s associated subject first, and falls back to TopicNameStrategy when the topic has no association. Existing applications keep working unchanged. The same producer and consumer code works for native, shared, and unassociated topics.

Clients still connect to the Kafka endpoint and the Schema Registry endpoint, each with its own credentials.

Supported client library versions

AssociatedNameStrategy is the default from the versions in this table. Earlier versions default to TopicNameStrategy and don’t use associations to find the subject.

Language	Client library	Earliest supported version
Java Client	Confluent serializers (`kafka-avro-serializer` and others)	8.0.x: 8.0.4 or later. 8.1.x: 8.1.2 or later. 8.2.x: 8.2.2 or later. 8.3.0 and later. Requires `kafka-clients` 4.1.0 or later.
Python	`confluent-kafka-python`	2.14.0
Go	`confluent-kafka-go`	2.14.0
Node.js	`@confluentinc/kafka-javascript` and `@confluentinc/schemaregistry`	1.9.0
.NET	`confluent-kafka-dotnet` and `libschemaregistry`	2.14.0 and 0.1.3

Caution

A producer on an earlier client library version doesn’t read associations. It builds the subject name from the topic name. For the topic customer-signups, it registers the schema under customer-signups-value and not the associated subject Customer. A producer that registers schemas automatically, the default, creates that second subject. You end up with two subjects for the same schema. To avoid this, upgrade the client library to a version in the preceding table.

Broker-side schema ID validation

Broker-side schema ID validation is an optional topic setting. When it’s on, the broker checks that the schema ID in each message is valid for the subject that the topic uses.

The broker uses AssociatedNameStrategy by default and finds the subject through the topic’s association. If the topic has no association, the broker falls back to TopicNameStrategy. You don’t need to configure anything on the topic.

Fallback strategy

If a topic has no association, AssociatedNameStrategy falls back to TopicNameStrategy by default. To choose the fallback, set subject.name.strategy.fallback.type on the client:

Value	Fallback
`TOPIC`	`TopicNameStrategy` (default)
`RECORD`	`RecordNameStrategy`
`TOPIC_RECORD`	`TopicRecordNameStrategy`
`NONE`	No fallback

Serialization fails with an error if the fallback is NONE and the topic has no association, or if the topic has more than one associated subject for the same key or value. Deserialization fails when the deserializer needs to resolve a subject. Examples include use.latest.version, use.latest.with.metadata, and migration rules. A deserializer that reads only the schema ID embedded in the message isn’t affected. Which deserializer settings resolve a subject varies by client library.

Authorization

Kafka and Schema Registry have separate endpoints, credentials, and authorization. You need an API key scoped to your Kafka cluster and a separate API key scoped to your Schema Registry cluster. These operations use the DeveloperWrite and DeveloperRead roles.

Operation	Kafka credential	Schema Registry credential	Kafka authorization	Schema Registry authorization
Create a topic with a native schema	Required	Not required	`DeveloperWrite` on the topic	`DeveloperWrite` on the subject
Register a shared schema	Not required	Required	Not required	`DeveloperWrite` on the subject
Associate a topic with a shared subject	Required	Not required	`DeveloperWrite` on the topic	`DeveloperWrite` on the subject
Produce	Required	Required	`DeveloperWrite` on the topic	`DeveloperRead` on the subject
Consume	Required	Required	`DeveloperRead` on the topic	`DeveloperRead` on the subject
Look up associations by subject	Not required	Required	Not required	Read access to the subject

Limits

Confluent Cloud limits how many Kafka administration requests that carry schema associations a Dedicated cluster accepts. The limits apply per cluster and are the same at every CKU size. Administration requests that don’t involve associations aren’t affected and continue to use the standard Kafka request limits.

If you exceed a limit, Confluent Cloud can throttle or reject your requests. Retry with exponential backoff.

Operation	Requests per second	Requests per five minutes
`CreateTopics` with an association	8	2,400
`IncrementalAlterConfigs` with an association	8	2,400
`DeleteTopics` on topics that have associations	8	2,400
`DescribeConfigs` that reads associations	50	15,000

The per-second value is a ceiling at every moment. The five-minute value is the per-second value multiplied by 300, and it doesn’t allow bursts over the per-second rate. For example, sending 2,400 CreateTopics requests in the first minute exceeds the limit even though the total is within 2,400 per five minutes.

Best practices

Send one topic per request. Batching topics with associations in a single request reduces performance. To create many topics, send one request per topic and stay within the per-second limit in the Limits table.
Expect lower throughput with larger requests. Confluent measured the limits with one association of about 20 KB per request. A topic can have two associations, key and value. Two associations or larger schemas lower the rate you can sustain. The total payload of a create or alter request can be up to 1 MB.
Keep Kafka and Schema Registry in the same region. Confluent measured the limits with Kafka and Schema Registry on the same continent. Using the same region improves performance.
The limits are maximums, not guarantees. Schema Registry also limits the requests it accepts for each environment. Every cluster in the environment shares that limit with all other Schema Registry traffic, such as clients registering and looking up schemas. If other traffic uses that capacity, a cluster might see throttling below the rates in the Limits table.
The limits don’t increase with cluster size. Larger clusters might see less throttling, but the numbers in the preceding table are the most Confluent supports at any size.

Replication across clusters

Cluster Linking doesn’t replicate associations. To replicate a topic that has an association, use Cluster Linking to mirror the topic and Schema Linking to replicate its schema. Then create the associations that exist on the source topic manually on the destination topic.