> ## Documentation Index
> Fetch the complete documentation index at: https://docs.streamnative.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Kafka Schema Registry Tutorial

> Register a schema, produce and consume typed records, then evolve the schema—watch an incompatible change fail and fix it.

This tutorial takes an Avro schema through its full life: register it, produce and consume records
with it, then change it—first in a way that fails the compatibility check, then in a way that
passes. The failure is the point. Seeing the registry reject a change is what makes the compatibility
rules concrete.

Allow about 20 minutes.

## Prerequisites

* A Kafka cluster, or a Pulsar cluster with the Kafka protocol enabled, on StreamNative Cloud.
* An API key for a service account with the
  [`schema-manager`](/cloud/security/access/rbac/manage-rbac-roles#schema-manager) role, plus produce
  and consume permissions on the topic.
* Java 17 or later and Maven, for the producer and consumer.
* `curl` and `jq`.

Set up your shell:

```shell theme={null}
export SR_URL="https://<schema-registry-url>"
export BOOTSTRAP="<cluster-name>-<instance-name>.<org-name>.streamnative.cloud:9093"
export API_KEY="<your-api-key>"
```

See [Connect](/kafka/governance/sr/connect) for how to find these.

## Step 1: Create a topic

```shell theme={null}
snctl kafka admin topics create payments --partitions 3
```

## Step 2: Define and register a schema

Create `payment.avsc`:

```json theme={null}
{
  "type": "record",
  "name": "Payment",
  "namespace": "com.acme",
  "fields": [
    {"name": "id", "type": "string"},
    {"name": "amount", "type": "double"}
  ]
}
```

Register it under the `payments-value` subject:

```shell theme={null}
curl -u "any-user:$API_KEY" \
  -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data "{\"schemaType\":\"AVRO\",\"schema\": $(jq -Rs . < payment.avsc)}" \
  "$SR_URL/subjects/payments-value/versions"
```

```json theme={null}
{"id":1}
```

That ID is what every record will carry. Confirm the subject exists:

```shell theme={null}
curl -u "any-user:$API_KEY" "$SR_URL/subjects/payments-value/versions"
```

```json theme={null}
[1]
```

## Step 3: Build a producer

Add the dependencies:

```xml theme={null}
<dependencies>
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.1</version>
  </dependency>
  <dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>7.5.0</version>
  </dependency>
</dependencies>

<repositories>
  <repository>
    <id>confluent</id>
    <url>https://packages.confluent.io/maven/</url>
  </repository>
</repositories>
```

Configure and send:

```java theme={null}
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, System.getenv("BOOTSTRAP"));
props.put("security.protocol", "SASL_SSL");
props.put("sasl.mechanism", "PLAIN");
props.put("sasl.jaas.config",
    "org.apache.kafka.common.security.plain.PlainLoginModule required "
  + "username=\"public/default\" password=\"token:" + System.getenv("API_KEY") + "\";");

props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
props.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, System.getenv("SR_URL"));
props.put(KafkaAvroSerializerConfig.BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
props.put(KafkaAvroSerializerConfig.USER_INFO_CONFIG, "any-user:" + System.getenv("API_KEY"));

Schema schema = new Schema.Parser().parse(new File("payment.avsc"));

try (Producer<String, GenericRecord> producer = new KafkaProducer<>(props)) {
    for (int i = 0; i < 5; i++) {
        GenericRecord payment = new GenericData.Record(schema);
        payment.put("id", "payment-" + i);
        payment.put("amount", 100.0 + i);
        producer.send(new ProducerRecord<>("payments", "payment-" + i, payment));
    }
}
```

## Step 4: Consume the records

```java theme={null}
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "payments-tutorial");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

try (Consumer<String, GenericRecord> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(List.of("payments"));
    for (ConsumerRecord<String, GenericRecord> record : consumer.poll(Duration.ofSeconds(10))) {
        System.out.printf("%s -> %s%n", record.key(), record.value());
    }
}
```

The consumer never sees `payment.avsc`. It reads the schema ID from each record's prefix and fetches
the schema from the registry—which is the whole point of the registry.

## Step 5: Make an incompatible change

Add a required field with no default. Create `payment-v2-bad.avsc`:

```json theme={null}
{
  "type": "record",
  "name": "Payment",
  "namespace": "com.acme",
  "fields": [
    {"name": "id", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "currency", "type": "string"}
  ]
}
```

Test it before registering:

```shell theme={null}
curl -u "any-user:$API_KEY" \
  -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data "{\"schema\": $(jq -Rs . < payment-v2-bad.avsc)}" \
  "$SR_URL/compatibility/subjects/payments-value/versions/latest?verbose=true"
```

```json theme={null}
{"is_compatible":false,"messages":["READER_FIELD_MISSING_DEFAULT_VALUE: currency"]}
```

The subject's default mode is `BACKWARD`, which requires that a consumer on the new schema can read
data written under the old one. The five records you produced have no `currency`, and the new schema
gives the reader nothing to substitute—so the change is rejected.

Try to register it anyway:

```shell theme={null}
curl -u "any-user:$API_KEY" \
  -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data "{\"schemaType\":\"AVRO\",\"schema\": $(jq -Rs . < payment-v2-bad.avsc)}" \
  "$SR_URL/subjects/payments-value/versions"
```

```json theme={null}
{"error_code":409,"message":"Schema being registered is incompatible with an earlier schema"}
```

HTTP 409. The registry stopped a change that would have broken every consumer of the existing data.

## Step 6: Fix it

Give the field a default. Create `payment-v2.avsc`:

```json theme={null}
{
  "type": "record",
  "name": "Payment",
  "namespace": "com.acme",
  "fields": [
    {"name": "id", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "currency", "type": "string", "default": "USD"}
  ]
}
```

```shell theme={null}
curl -u "any-user:$API_KEY" \
  -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data "{\"schemaType\":\"AVRO\",\"schema\": $(jq -Rs . < payment-v2.avsc)}" \
  "$SR_URL/subjects/payments-value/versions"
```

```json theme={null}
{"id":2}
```

A new schema ID and a new version:

```shell theme={null}
curl -u "any-user:$API_KEY" "$SR_URL/subjects/payments-value/versions"
```

```json theme={null}
[1,2]
```

Now a consumer on v2 reading a v1 record gets `currency = "USD"`—the default fills the gap. That
single word is the difference between a safe change and a broken pipeline.

## Step 7: Change the compatibility mode

Sometimes you genuinely need a breaking change. Loosen the mode deliberately rather than working
around the check:

```shell theme={null}
curl -u "any-user:$API_KEY" \
  -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility":"NONE"}' \
  "$SR_URL/config/payments-value"
```

Now `payment-v2-bad.avsc` registers. Set the mode back when you're done:

```shell theme={null}
curl -u "any-user:$API_KEY" \
  -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility":"BACKWARD"}' \
  "$SR_URL/config/payments-value"
```

<Warning>
  `NONE` disables the guarantee, it doesn't make old consumers able to read new data. Use it only when
  you control every consumer and can coordinate the rollout.
</Warning>

<Note title="Read the per-subject config">
  `GET /config/payments-value` reports the real value. The global `GET /config` always returns `NONE`
  regardless of what's set—see
  [Confluent API compatibility](/kafka/governance/sr/reference/confluent-compatibility#get-config-always-reports-none).
</Note>

## Clean up

```shell theme={null}
# Soft delete, then hard delete
curl -u "any-user:$API_KEY" -X DELETE "$SR_URL/subjects/payments-value"
curl -u "any-user:$API_KEY" -X DELETE "$SR_URL/subjects/payments-value?permanent=true"

snctl kafka admin topics delete payments
```

## What to take away

* A schema ID travels in each record; the consumer resolves it from the registry rather than being
  told the schema out of band.
* `BACKWARD`, the default, means new-schema consumers must be able to read old data—so **upgrade
  consumers before producers**.
* Defaults are what make a field addition compatible in Avro.
* Test with `/compatibility/...` in CI, so an incompatible change fails the build instead of the
  deploy.

## Next steps

<CardGroup cols={2}>
  <Card title="Evolution and compatibility" icon="code-compare" href="/kafka/governance/sr/fundamentals/schema-evolution">
    Every mode, and which side to upgrade first.
  </Card>

  <Card title="Manage schemas" icon="sliders" href="/kafka/governance/sr/manage/manage-schemas">
    A registration workflow that holds up in production.
  </Card>
</CardGroup>