Kafka Schema Registry Tutorial

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

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 for how to find these.

Step 1: Create a topic

snctl kafka admin topics create payments --partitions 3

Step 2: Define and register a schema

Create payment.avsc:

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

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"

{"id":1}

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

curl -u "any-user:$API_KEY" "$SR_URL/subjects/payments-value/versions"

[1]

Step 3: Build a producer

Add the dependencies:

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

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

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:

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

Test it before registering:

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"

{"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:

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"

{"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:

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

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"

{"id":2}

A new schema ID and a new version:

curl -u "any-user:$API_KEY" "$SR_URL/subjects/payments-value/versions"

[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:

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:

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

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.

GET /config/payments-value reports the real value. The global GET /config always returns NONE regardless of what’s set—see Confluent API compatibility.

Clean up

# 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.

Introduction

Get Started

Kafka Clusters

Process

Lakehouse

Governance

Migrate

Kafka Clients

Prerequisites

Step 1: Create a topic

Step 2: Define and register a schema

Step 3: Build a producer

Step 4: Consume the records

Step 5: Make an incompatible change

Step 6: Fix it

Step 7: Change the compatibility mode

Clean up

What to take away

Next steps

Evolution and compatibility

Manage schemas

​Prerequisites

​Step 1: Create a topic

​Step 2: Define and register a schema

​Step 3: Build a producer

​Step 4: Consume the records

​Step 5: Make an incompatible change

​Step 6: Fix it

​Step 7: Change the compatibility mode

​Clean up

​What to take away

​Next steps

Evolution and compatibility

Manage schemas

Prerequisites

Step 1: Create a topic

Step 2: Define and register a schema

Step 3: Build a producer

Step 4: Consume the records

Step 5: Make an incompatible change

Step 6: Fix it

Step 7: Change the compatibility mode

Clean up

What to take away

Next steps