Schema Registry und Avro/Protobuf in Kafka mit Spring Boot

1. Überblick

Wenn mehrere Services ein und dasselbe Kafka-Topic lesen und schreiben, ist die größte Gefahr das Nachrichtenformat. Sobald ein Producer einen Feldnamen in JSON ändert, können alle nachgelagerten Consumer sofort kaputt gehen. Confluent Schema Registry löst dieses Problem. Sie speichert Schemas zentral, verknüpft jede Nachricht mit einer Schema-ID und blockiert inkompatible Änderungen.

Dieser Beitrag führt dich durch das Aufsetzen von Schema Registry, das Versenden und Empfangen von Nachrichten mit Avro und Protobuf in Spring Boot 3.x, das Konfigurieren von Kompatibilität und das Schreiben von Tests ohne echten Kafka.

2. So funktioniert Schema Registry

Schema Registry ist ein eigenständiger REST-Service, der neben dem Kafka-Cluster läuft. Intern speichert er Schemas im Topic _schemas. Jedes Schema gehört zu einem Subject. Nach Standard (TopicNameStrategy) hat das Subject des Values im Topic orders den Namen orders-value.

Wenn ein Producer eine Nachricht versendet, führt der Serializer folgende Schritte durch:

Das Schema wird bei der Registry registriert oder nachgeschlagen, um die Schema-ID zu erhalten. Das Ergebnis wird gecacht.
Der Payload wird nach dem Wire-Format von Confluent geschrieben:

| Byte 0: Magic Byte (0) | Byte 1-4: Schema-ID (int) | Byte 5+: Binäre Avro/Protobuf-Daten |

Der Consumer liest die ersten 5 Bytes, holt sich die Schema-ID, lädt das Schema einmalig und cacht es, dann dekodiert er den Payload. Die Nachricht muss nicht das gesamte Schema mitbringen, daher ist sie viel kleiner als JSON.

3. Lokale Umgebung aufsetzen

Nutze Docker Compose mit Kafka im KRaft-Modus und Schema Registry:

services:
  kafka:
    image: confluentinc/cp-kafka:7.7.1
    ports: ["9092:9092"]
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:29093
      KAFKA_LISTENERS: PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,EXTERNAL://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

  schema-registry:
    image: confluentinc/cp-schema-registry:7.7.1
    depends_on: [kafka]
    ports: ["8081:8081"]
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKA_STORE_BOOTSTRAP_SERVERS: kafka:29092
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

Nach dem Ausführen von docker compose up -d prüfst du die Registry:

curl -s http://localhost:8081/subjects

[]

Das Ergebnis ist eine leere Liste, da noch keine Subjects registriert wurden.

4. Avro mit Spring Boot

4.1. Dependencies und Plugin

Die Serializer von Confluent sind nicht auf Maven Central verfügbar. Du musst das Repository von Confluent hinzufügen:

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

<dependencies>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-avro-serializer</artifactId>
        <version>7.7.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>1.12.0</version>
    </dependency>
</dependencies>

Das Plugin avro-maven-plugin erzeugt Java-Klassen aus .avsc-Dateien:

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.12.0</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>
                <stringType>String</stringType>
            </configuration>
        </execution>
    </executions>
</plugin>

Die Option stringType=String sorgt dafür, dass Zeichenketten-Felder als java.lang.String statt als CharSequence erzeugt werden. Dadurch werden Vergleiche im Code und in Tests vereinfacht.

4.2. Schema definieren

Erstelle die Datei src/main/avro/Order.avsc:

{
  "type": "record",
  "name": "Order",
  "namespace": "dev.nonla.order.avro",
  "fields": [
    { "name": "id", "type": "string" },
    { "name": "customerId", "type": "string" },
    { "name": "amount", "type": "double" }
  ]
}

Führe mvn generate-sources aus. Das Plugin erzeugt die Klasse dev.nonla.order.avro.Order mit Builder und eingebettetem Schema.

4.3. Producer und Consumer konfigurieren

spring:
  kafka:
    bootstrap-servers: localhost:9092
    properties:
      schema.registry.url: http://localhost:8081
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
    consumer:
      group-id: shipping-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      properties:
        specific.avro.reader: true

Die Eigenschaft schema.registry.url wird unter spring.kafka.properties gesetzt, damit sie sowohl auf Producer als auch Consumer angewendet wird. Die Eigenschaft specific.avro.reader: true ist sehr wichtig. Ohne sie empfängt der Consumer GenericRecord statt der erzeugten Klasse Order.

4.4. Nachrichten versenden und empfangen

@Service
public class OrderPublisher {

    private final KafkaTemplate<String, Order> kafkaTemplate;

    public OrderPublisher(KafkaTemplate<String, Order> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publish(String orderId, String customerId, double amount) {
        Order order = Order.newBuilder()
                .setId(orderId)
                .setCustomerId(customerId)
                .setAmount(amount)
                .build();
        kafkaTemplate.send("orders", order.getId(), order);
    }
}

@Component
public class ShippingListener {

    private static final Logger log = LoggerFactory.getLogger(ShippingListener.class);

    @KafkaListener(topics = "orders")
    public void onOrder(Order order) {
        log.info("Versand wird vorbereitet für Bestellung {} des Kunden {}, Betrag {}",
                order.getId(), order.getCustomerId(), order.getAmount());
    }
}

Beim ersten Versand registriert KafkaAvroSerializer das Schema automatisch zum Subject orders-value. Du kannst das überprüfen:

curl -s http://localhost:8081/subjects/orders-value/versions/latest

{"subject":"orders-value","version":1,"id":1,"schema":"{\"type\":\"record\",\"name\":\"Order\",...}"}

Ab jetzt trägt die Nachricht im Topic nur die Schema-ID 1, nicht das gesamte Schema.

5. Protobuf mit Spring Boot

Die Schritte mit Protobuf sind fast identisch mit Avro, nur unterschiedliche Serializer und Code-Generierung. Füge diese Dependency hinzu:

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-protobuf-serializer</artifactId>
    <version>7.7.1</version>
</dependency>

Definiere src/main/proto/order_event.proto:

syntax = "proto3";

package dev.nonla.order;

option java_package = "dev.nonla.order.proto";
option java_multiple_files = true;

message OrderEvent {
  string id = 1;
  string customer_id = 2;
  double amount = 3;
}

Du erzeugst Java-Klassen mit einem protoc-Plugin für Maven, zum Beispiel io.github.ascopes:protobuf-maven-plugin. Dann konfigurierst du den Serializer:

spring:
  kafka:
    producer:
      value-serializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer
    consumer:
      value-deserializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer
      properties:
        specific.protobuf.value.type: dev.nonla.order.proto.OrderEvent

Die Eigenschaft specific.protobuf.value.type teilt dem Deserializer die Zielklasse mit. Ohne sie empfängst du DynamicMessage.

Welches Format solltest du wählen?

Avro eignet sich für das Kafka-Ökosystem und Data-Pipelines wie Kafka Connect oder ksqlDB. Schema ist JSON-Format, Evolutionsregeln basieren auf Feldnamen und Standardwerten.
Protobuf eignet sich, wenn deine Organisation bereits gRPC nutzt. Du kannst vorhandene .proto-Dateien wiederverwenden. Evolutionsregeln basieren auf Feldnummern.

6. Schema-Evolution und Kompatibilität

6.1. Kompatibilitätsstufen

Jedes Subject hat eine Kompatibilitätsstufe. Der Standard ist BACKWARD.

Stufe	Bedeutung	Sichere Reihenfolge beim Deployment
`BACKWARD`	Neues Schema kann alte Daten lesen	Consumer zuerst, Producer danach
`FORWARD`	Altes Schema kann neue Daten lesen	Producer zuerst, Consumer danach
`FULL`	Funktioniert in beide Richtungen	Beliebige Reihenfolge
`*_TRANSITIVE`	Prüfung gegen alle alten Versionen, nicht nur die vorherige	Wie oben

Du stellst die Stufe für jedes Subject über die REST API ein:

curl -s -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility": "FULL_TRANSITIVE"}' \
  http://localhost:8081/config/orders-value

6.2. Eine gültige Änderung

Angenommen, du möchtest ein Feld currency zu Order hinzufügen. Bei Avro muss das neue Feld einen Standardwert haben, damit das Schema FULL erreicht:

{ "name": "currency", "type": "string", "default": "VND" }

Bevor du den Code mergest, solltest du das neue Schema mit der Registry überprüfen:

curl -s -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"dev.nonla.order.avro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"customerId\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"currency\",\"type\":\"string\",\"default\":\"VND\"}]}"}' \
  http://localhost:8081/compatibility/subjects/orders-value/versions/latest

{"is_compatible":true}

Wenn du "default": "VND" weglässt, wird false zurückgegeben. Der Producer erhält dann einen 409 Conflict-Fehler beim Versuch, das Schema zu registrieren.

Bei Protobuf ist die entsprechende Regel: Ändere oder wiederhole niemals eine Feldnummer. Wenn du ein Feld löschst, markiere es mit reserved, damit niemand die Nummer wiederverwenden kann:

message OrderEvent {
  reserved 2;
  reserved "customer_id";
  string id = 1;
  double amount = 3;
  string currency = 4;
}

7. Tests mit Mock Schema Registry schreiben

Die Serializer von Confluent unterstützen URLs wie mock://<scope>. Mit dieser URL nutzt der Serializer eine simulierte Registry im Speicher. Du kannst Serialisierung testen, ohne Docker zu brauchen:

import static org.assertj.core.api.Assertions.assertThat;

class OrderAvroSerdeTest {

    private static final Map<String, Object> CONFIG = Map.of(
            "schema.registry.url", "mock://order-serde-test",
            "specific.avro.reader", true);

    @Test
    void shouldSerializeWithWireFormatAndDeserializeBack() {
        try (var serializer = new KafkaAvroSerializer();
             var deserializer = new KafkaAvroDeserializer()) {
            serializer.configure(CONFIG, false);
            deserializer.configure(CONFIG, false);

            Order order = Order.newBuilder()
                    .setId("ORD-1001")
                    .setCustomerId("CUS-42")
                    .setAmount(250_000.0)
                    .build();

            byte[] bytes = serializer.serialize("orders", order);

            assertThat(bytes[0]).isZero();                       // Magic Byte
            assertThat(ByteBuffer.wrap(bytes, 1, 4).getInt()).isPositive(); // Schema-ID

            Object result = deserializer.deserialize("orders", bytes);
            assertThat(result).isInstanceOf(Order.class).isEqualTo(order);
        }
    }
}

Dieser Test prüft drei Dinge:

Das erste Byte ist das Magic Byte 0.
Die nächsten vier Bytes bilden eine gültige Schema-ID.
Der Deserializer gibt die korrekte Klasse Order zurück, nicht GenericRecord. Das bestätigt, dass specific.avro.reader aktiv ist.

Für vollständige Integrationstests kannst du Testcontainers nutzen, um Kafka mit Schema Registry zu starten. Tests mit mock:// sind jedoch schneller und reichen aus, um die meisten Konfigurationsfehler zu finden.

Hinweis: Serializer, die denselben mock://-Scope nutzen, teilen sich eine simulated Registry. Jede Test-Klasse sollte ihren eigenen Scope verwenden, um gegenseitige Beeinflussungen zu vermeiden.

8. Häufige Fehler und wann man Registry nicht einsetzen sollte

Poison Pill. Eine Nachricht mit falschem Wire-Format lässt KafkaAvroDeserializer eine SerializationException werfen, bevor sie den Listener erreicht. Der Consumer kann bei diesem Offset steckenbleiben. Du solltest den Deserializer mit ErrorHandlingDeserializer umhüllen:

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: io.confluent.kafka.serializers.KafkaAvroDeserializer

Automatische Schema-Registrierung in Production. Standardmäßig ist auto.register.schemas=true, d. h. jeder Producer kann neue Schemas registrieren. In Production solltest du auto.register.schemas: false und use.latest.version: true setzen. Schemas sollten nur über deine CI/CD-Pipeline nach Kompatibilitätsprüfung registriert werden.

Unbeabsichtigtes Empfangen von GenericRecord oder DynamicMessage. Der Grund ist meist, dass du specific.avro.reader oder specific.protobuf.value.type vergessen hast. Die Folge ist ein ClassCastException im Listener.

Namespace oder Record-Name in Avro ändern. Das ist ein völlig anderes Schema. Der Specific Reader findet keine passende Klasse. Behandle den Namespace als unveränderlichen Teil des Vertrags.

Kompatibilität auf NONE setzen, um es zu vereinfachen. Das macht den gesamten Sinn von Registry zunichte. Nutze NONE nur für experimentelle Topics.

Hinweis: Schema Registry ist eine Laufzeit-Abhängigkeit. Der Serializer cacht Schemas, daher hat ein temporär ausgefallenes Registry selten Auswirkungen auf bekannte Schemas. Ein neuer Producer oder ein unbekanntes Schema führt aber sofort zu einem Fehler. Betreibe Registry in Production in Multi-Instance-Modus.

Wann solltest du Registry nicht einsetzen:

Kleines System, nur ein Team besitzt Producer und Consumer, wenig Änderungen. JSON mit Springs JsonSerializer ist viel einfacher.
Nachrichten müssen direkt von Clients gelesen werden, die das Wire-Format von Confluent nicht verstehen, zum Beispiel ein Tool eines Drittanbieters, das nur reines JSON liest.
Du kannst keinen zusätzlichen Infrastruktur-Service betreiben und hast auch keinen verwalteten Registry.

9. Fazit

Schema Registry verwandelt Kafka-Message-Formate in einen versionierten Vertrag mit automatischer Validierung. Du hast gelernt, wie du:

Registry mit Docker aufsetzen kannst.
Nachrichten mit KafkaAvroSerializer und KafkaProtobufSerializer in Spring Boot 3 versendest und empfängst.
Das Wire-Format mit Magic Byte und Schema-ID verstehst.
Kompatibilitätsstufen setzt und Schemas sicher weiterentwickelst: neue Felder mit Standardwerten bei Avro, reserved bei Protobuf.
Schnelle Tests mit mock://-Registry schreibst.

Beim Einsatz in Production deaktivierst du die automatische Schema-Registrierung, integrierst Kompatibilitätsprüfungen in CI und hüllst den Deserializer immer mit ErrorHandlingDeserializer ein. So wird eine fehlerhafte Schema-Änderung bereits beim Review abgefangen, nicht wenn Consumers mitten in der Nacht ausfallen.