60-Day Kafka 4 Learning Plan · Week 4 — Day 23 of 60

---

60-Day Kafka 4 Learning Plan · Week 4 — Schema & Serialization Sources: Kafka: The Definitive Guide Ch.3 · docs.confluent.io/schema-registry

Goal

Stand up Confluent Schema Registry alongside Kafka 4, understand the publish/consume flow with Avro, configure KafkaAvroSerializer in Spring Boot, learn how subject naming strategies control schema versioning, and understand how to run, secure, and test against Schema Registry in production.

1. What Schema Registry does

labs-api (producer)
  → register schema with Registry (first publish only)
  → receive schema_id back
  → serialize: [0x00 magic][schema_id 4B][avro binary]
  → publish to Kafka 4

Kafka 4 topic (stores opaque bytes)

labs-socket (consumer)
  → read message from Kafka
  → extract schema_id from bytes 1–4
  → fetch schema from Registry (cached after first time)
  → deserialize bytes → OrderEvent POJO

Schema Registry is a REST service — producers and consumers talk to it over HTTP. Schemas are cached client-side after the first fetch, so there’s no round-trip overhead per message in steady state.

Subjects: Registry organises schemas by subject — one version history per subject. Default naming: {topic}-value and {topic}-key.

2. Docker Compose — add Schema Registry

# docker-compose.yml (add alongside existing kafka service)
services:
  kafka:
    image: apache/kafka:4.0.0
    # ... (existing config from Day 21)

  schema-registry:
    image: confluentinc/cp-schema-registry:7.6.1
    ports: ["8081:8081"]
    depends_on: [kafka]
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

docker compose up -d
# Verify Registry is healthy
curl http://localhost:8081/subjects
# → []  (empty — no schemas registered yet)

3. Schema Registry REST API

Register a schema

curl -X POST http://localhost:8081/subjects/labs.events-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{
    "schema": "{\"type\":\"record\",\"name\":\"OrderEvent\",\"namespace\":\"com.labs.events\",\"fields\":[{\"name\":\"orderId\",\"type\":\"string\"},{\"name\":\"userId\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"status\",\"type\":\"string\"}]}"
  }'
# → {"id": 1}

Useful REST endpoints

# List all subjects
curl http://localhost:8081/subjects

# Get all versions for a subject
curl http://localhost:8081/subjects/labs.events-value/versions

# Get schema by id
curl http://localhost:8081/schemas/ids/1

# Get latest schema for a subject
curl http://localhost:8081/subjects/labs.events-value/versions/latest

# Delete a subject (careful in production!)
curl -X DELETE http://localhost:8081/subjects/labs.events-value

4. Maven dependencies (both services)

<!-- Confluent Maven repository required -->
<repositories>
  <repository>
    <id>confluent</id>
    <url>https://packages.confluent.io/maven/</url>
  </repository>
</repositories>

<dependencies>
  <!-- Avro runtime -->
  <dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
    <version>1.11.3</version>
  </dependency>

  <!-- Confluent Avro serializers (includes Schema Registry client) -->
  <dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>7.6.1</version>
  </dependency>

  <!-- Avro Maven plugin for code generation from .avsc -->
  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro-maven-plugin</artifactId>
        <version>1.11.3</version>
        <executions>
          <execution>
            <phase>generate-sources</phase>
            <goals><goal>schema</goal></goals>
            <configuration>
              <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
              <outputDirectory>${project.build.directory}/generated-sources/avro/</outputDirectory>
            </configuration>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>
</dependencies>

5. Spring Boot application.yml — Avro config

# application.yml (shared — both labs-api and labs-socket)
spring:
  kafka:
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
      acks: all
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      auto-offset-reset: earliest
    properties:
      schema.registry.url: http://localhost:8081
      specific.avro.reader: true    # deserialize to generated POJO, not GenericRecord
      # Optional: auto-register schemas on producer side (disable in prod, use CI/CD pipeline — see §10)
      auto.register.schemas: true

specific.avro.reader: true vs false

6. Subject naming strategies

Configure in application.yml:

spring:
  kafka:
    properties:
      value.subject.name.strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy

For most projects, the default TopicNameStrategy is correct — one schema per Kafka topic.

7. Verify end-to-end

// labs-api EventProducer — now using generated Avro POJO
@Service @RequiredArgsConstructor
public class EventProducer {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public void publish(OrderEvent event) {
        kafkaTemplate.send("labs.events", event.getUserId().toString(), event);
    }
}

# After publishing an event, check the Registry received the schema
curl http://localhost:8081/subjects
# → ["labs.events-value"]

curl http://localhost:8081/subjects/labs.events-value/versions/latest | jq .
# → {"subject":"labs.events-value","version":1,"id":1,"schema":"{...}"}

8. Schema Registry HA — it’s not a database, it’s a Kafka consumer

Schema Registry doesn’t have its own persistent storage — it stores every schema as a record in an internal Kafka topic (_schemas, compacted) and rebuilds its in-memory schema cache by consuming that topic on startup. This has real implications:

• Multiple Registry instances behind a load balancer provide HA, but only one is the active writer at a time (leader election coordinated via the same _schemas topic) — reads can be served by any instance, but schema registration requests get forwarded to the current leader.
• The _schemas topic needs the same durability treatment as any critical topic — replication.factor=3, min.insync.replicas=2 (Day 10 §3). Losing that topic’s data means losing every registered schema’s history, which breaks every consumer that needs to resolve a schema ID it hasn’t cached yet.
• A fresh Registry instance is not “empty” once _schemas exists — on startup it replays the entire topic to rebuild state, so recovery time scales with schema registration history size, similar in spirit to the KRaft snapshot/replay pattern from Day 8.

schema-registry:
  environment:
    SCHEMA_REGISTRY_KAFKASTORE_TOPIC: _schemas
    SCHEMA_REGISTRY_KAFKASTORE_TOPIC_REPLICATION_FACTOR: 3   # production: never 1

9. Securing Schema Registry

The compose setup in §2 has no authentication — any client that can reach port 8081 can register, modify, or delete schemas, including deleting a subject entirely (§3’s delete endpoint).

schema-registry:
  environment:
    SCHEMA_REGISTRY_AUTHENTICATION_METHOD: BASIC
    SCHEMA_REGISTRY_AUTHENTICATION_ROLES: admin,developer
    SCHEMA_REGISTRY_AUTHENTICATION_REALM: SchemaRegistry-Props

# Spring Boot client side
spring:
  kafka:
    properties:
      schema.registry.url: https://schema-registry:8081
      basic.auth.credentials.source: USER_INFO
      basic.auth.user.info: ${SCHEMA_REGISTRY_USER}:${SCHEMA_REGISTRY_PASSWORD}

Minimum bar for anything beyond local dev: authentication plus HTTPS. An unauthenticated Registry means anyone with network access can delete a subject’s schema history — a single curl -X DELETE away from every consumer relying on that subject breaking.

10. CI/CD schema registration — why auto.register.schemas=true is risky in production

With auto.register.schemas=true (§5), any producer deploy that includes a schema change registers it immediately on first publish, with no review step and no chance to catch an incompatible change before it reaches the Registry (and starts affecting consumers using latest).

Production pattern: disable auto-registration, and register schemas explicitly as a CI/CD pipeline step before the deploy that uses them — giving you a place to run compatibility checks and get review, the same way you’d review a database migration.

# CI pipeline step (pseudocode) — runs before deploying labs-api
- name: Validate and register schema
  run: |
    # Check compatibility against the currently registered version FIRST
    curl -X POST http://schema-registry:8081/compatibility/subjects/labs.events-value/versions/latest \
      -H "Content-Type: application/vnd.schemaregistry.v1+json" \
      -d "{\"schema\": $(cat src/main/avro/OrderEvent.avsc | jq -Rs .)}"
    # Only register if compatible (full compatibility rules in Day 24)
    curl -X POST http://schema-registry:8081/subjects/labs.events-value/versions \
      -H "Content-Type: application/vnd.schemaregistry.v1+json" \
      -d "{\"schema\": $(cat src/main/avro/OrderEvent.avsc | jq -Rs .)}"

spring:
  kafka:
    properties:
      auto.register.schemas: false   # production — registration happens in CI, not at runtime

What this buys you: a broken/incompatible schema change fails the CI pipeline before it ever reaches a running producer, instead of failing silently at runtime the first time that producer tries to publish — turning a potential production incident into a rejected pull request.

11. Testing serialization without a running Registry

Spinning up a real Schema Registry container for every unit test is slow and adds CI flakiness. MockSchemaRegistryClient provides an in-memory implementation for fast, isolated tests.

class OrderEventSerializationTest {

    @Test
    void serializesAndDeserializesWithMockRegistry() {
        SchemaRegistryClient mockRegistry = new MockSchemaRegistryClient();

        var serializer = new KafkaAvroSerializer(mockRegistry);
        var deserializer = new KafkaAvroDeserializer(mockRegistry);

        OrderEvent event = new OrderEvent("ord-1", "user-1", 99.99, Status.CONFIRMED);
        byte[] bytes = serializer.serialize("labs.events", event);

        OrderEvent result = (OrderEvent) deserializer.deserialize("labs.events", bytes);
        assertThat(result).isEqualTo(event);
    }
}

For full Spring Boot integration tests, @EmbeddedKafka (Day 15 §9) still needs a real or mock Registry reachable at the configured URL — MockSchemaRegistryClient is best reserved for focused serialization unit tests, not the full @KafkaListener integration path.

12. Monitoring Schema Registry

Practical habit: treat _schemas as a first-class production topic in your monitoring dashboards, not an implementation detail — every consumer’s ability to resolve a schema ID depends on it being healthy.

13. Common pitfalls

• auto.register.schemas=true in production — no review gate before an incompatible change reaches the Registry; see §10 for the CI/CD alternative
• Single Schema Registry instance with _schemas at RF=1 — same durability gap as any single-node Kafka setup; Registry’s HA story depends entirely on the underlying topic being properly replicated
• No authentication on the Registry — anyone with network access can delete a subject’s entire version history
• Assuming MockSchemaRegistryClient tests cover the full integration path — it verifies serialization logic, not @KafkaListener wiring, consumer group behavior, or a real Registry’s compatibility enforcement
• Choosing RecordNameStrategy/TopicRecordNameStrategy without a clear reason — the default TopicNameStrategy is correct for most projects; deviating adds subject-management complexity that should be a deliberate choice, not a default

Key Takeaways

• Schema Registry is a REST service storing Avro/Protobuf schemas keyed by subject
• Producers register schemas on first publish; consumers fetch by the 4-byte ID in each message
• Schemas are cached client-side — no Registry round-trip after the first fetch per schema ID
• KafkaAvroSerializer / KafkaAvroDeserializer replace StringSerializer in application.yml
• specific.avro.reader: true → deserializes to generated POJO instead of GenericRecord
• Default subject naming: {topic}-value (one schema version history per topic)
• Registry has no independent storage — it’s backed by the _schemas Kafka topic, so that topic needs production-grade replication like any other critical topic
• Disable auto.register.schemas in production; register schemas via a CI/CD compatibility-check step instead
• MockSchemaRegistryClient is for fast serialization unit tests, not a substitute for full integration testing

---

Support me through GitHub Sponsors.

Thank you for Reading !! See you in the next post.

Next

➡️ Day 24: Avro schemas — evolution & compatibility rules

Resources

• 📘 Kafka: The Definitive Guide — Chapter 3 (Serializers)
• 🌐 docs.confluent.io/platform/current/schema-registry
• 🌐 Confluent — Schema Registry security

👉 Link to Medium blog