# 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](#cloud-explicit-associations-clients). If the topic has no
association, it falls back to `TopicNameStrategy`. | | `TopicNameStrategy` | Legacy implicit | Derives the name from the topic | `-key` or `-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 | `-` | 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](#cloud-explicit-associations-clients). For details on the other legacy strategies, see [Subject name strategy](serdes-develop/index.md#sr-schemas-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](#cloud-explicit-associations-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, `-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, `-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](#cloud-explicit-associations-clients). - A Kafka API key and a Schema Registry API key. Kafka and Schema Registry keep separate endpoints, credentials, and authorization. See [Authorization](#cloud-explicit-associations-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 `-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: ```json { "type": "record", "name": "Order", "namespace": "com.example.avro", "fields": [ { "name": "orderId", "type": "string" }, { "name": "amount", "type": "double" } ] } ``` ### Cloud Console 1. Go to **Environment > Cluster > Topics > Add topic**. 2. 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**. 3. Set **Type** to **Avro** and paste the `Order` schema. Click **Validate**. 4. 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: ```bash 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: ```http 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: ```hcl 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`: ```java Properties props = new Properties(); props.put("bootstrap.servers", ""); try (AdminClient adminClient = AdminClient.create(props)) { NewTopic topic = new NewTopic("orders", 1, (short) 1); Map 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 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](../schemas-manage.md#cloud-schema-create). These examples use the `Customer` schema: ```json { "type": "record", "name": "Customer", "namespace": "com.example.avro", "fields": [ { "name": "customerId", "type": "string" }, { "name": "email", "type": "string" } ] } ``` ### Cloud Console 1. Go to **Environment > Schema Registry > Add schema**. 2. Set the subject to `Customer`, set the format to **Avro**, and paste the schema. 3. Click **Create**. ### Confluent CLI Save the `Customer` schema as `customer.avsc`, and then run this command to register it: ```bash confluent schema-registry schema create \ --subject Customer \ --schema customer.avsc \ --type avro ``` ### REST API Send this request to register the `Customer` schema: ```http 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: ```hcl 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](../develop/maven-plugin.md#sr-maven-plugin) 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: ```bash 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: ```http 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: ```hcl 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`: ```java Map 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: ```bash 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: ```http 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: ```hcl config = { "retention.ms" = "604800000" "confluent.value.association" = jsonencode({ subject = "Customer" }) } ``` to this: ```hcl config = { "retention.ms" = "604800000" } ``` Then apply the change. ### Java (AdminClient) Remove the association from the `customer-signups` topic through the `AdminClient`: ```java 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: ```http GET /associations/subjects/Customer ``` Example response: ```json [ { "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: ```http GET /associations/resources/{cluster_id}/orders ``` `{cluster_id}` is the Kafka cluster ID (`lkc-...`). Example response: ```json [ { "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 | ### 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](#cloud-explicit-associations-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](#cloud-explicit-associations-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.