Schema Registry et Avro/Protobuf dans Kafka avec Spring Boot

1. Vue d’ensemble

Quand plusieurs services lisent et écrivent le même topic Kafka, le risque majeur concerne le format des messages. Il suffit qu’un producer change le nom d’un champ dans JSON pour que les consumers en aval se cassent immédiatement. Confluent Schema Registry résout ce problème. Il stocke les schémas de manière centralisée, attache chaque message à un schema ID et bloque les changements qui cassent la compatibilité.

Cet article vous guide pour mettre en place Schema Registry, envoyer et recevoir des messages avec Avro et Protobuf dans Spring Boot 3.x, configurer la compatibilité, et écrire des tests sans Kafka réel.

2. Comment Schema Registry fonctionne

Schema Registry est un service REST distinct, situé à côté du cluster Kafka. À l’intérieur, il stocke les schémas dans un topic _schemas. Chaque schéma appartient à un subject. Par défaut (TopicNameStrategy), le subject de la value dans le topic orders s’appelle orders-value.

Quand un producer envoie un message, le sérialiseur fait les étapes suivantes :

Enregistrer ou rechercher le schéma sur Registry pour obtenir le schema ID. Le résultat est mis en cache.
Écrire la charge utile selon le wire format de Confluent :

| byte 0: magic byte (0) | byte 1-4: schema ID (int) | byte 5..: données Avro/Protobuf binaires |

Le consumer lit les 5 premiers octets, obtient le schema ID, charge le schéma une seule fois puis le met en cache, puis il décode la charge utile. Le message n’a pas besoin de porter l’intégralité du schéma, donc la taille est beaucoup plus petite que JSON.

3. Mettre en place l’environnement local

Utilisez Docker Compose avec Kafka en mode KRaft et 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

Après avoir exécuté docker compose up -d, vérifiez Registry :

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

[]

Le résultat est une liste vide car aucun subject n’a été enregistré pour le moment.

4. Avro avec Spring Boot

4.1. Dépendance et plugin

Les sérialiseurs de Confluent ne sont pas sur Maven Central. Vous devez déclarer le repository de Confluent :

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

Le plugin avro-maven-plugin génère des classes Java à partir des fichiers .avsc :

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

L’option stringType=String fait en sorte que les champs de chaîne soient générés en tant que java.lang.String au lieu de CharSequence. Cela rend la comparaison dans le code et dans les tests plus simple.

4.2. Définition du schéma

Créez le fichier 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" }
  ]
}

Exécutez mvn generate-sources. Le plugin générera la classe dev.nonla.order.avro.Order, avec un builder et le schéma intégré dans la classe.

4.3. Configuration du producer et du consumer

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

La propriété schema.registry.url est définie sous spring.kafka.properties donc elle s’applique à la fois au producer et au consumer. La propriété specific.avro.reader: true est très importante. Sans elle, le consumer reçoit GenericRecord au lieu de la classe Order générée.

4.4. Envoyer et recevoir des messages

@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("Préparation de la livraison de la commande {} pour le client {}, valeur {}",
                order.getId(), order.getCustomerId(), order.getAmount());
    }
}

Lors du premier envoi, KafkaAvroSerializer enregistre automatiquement le schéma dans le subject orders-value. Vous pouvez vérifier à nouveau :

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

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

À ce moment, le message sur le topic ne porte que le schema ID 1, pas l’intégralité du schéma.

5. Protobuf avec Spring Boot

Les étapes avec Protobuf sont presque identiques à celles d’Avro, seulement le sérialiseur et la génération de code changent. Ajoutez la dépendance suivante :

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

Définissez 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;
}

Vous générez la classe Java avec un plugin protoc pour Maven, par exemple io.github.ascopes:protobuf-maven-plugin. Ensuite, configurez le sérialiseur :

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

La propriété specific.protobuf.value.type indique au désérialiseur la classe cible. Sans elle, vous recevez DynamicMessage.

Lequel choisir ?

Avro convient bien à l’écosystème Kafka et aux pipelines de données comme Kafka Connect ou ksqlDB. Le schéma est en JSON, les règles d’évolution sont basées sur le nom du champ et les valeurs par défaut.
Protobuf convient quand votre organisation utilise déjà gRPC. Vous pouvez réutiliser les fichiers .proto existants. Les règles d’évolution reposent sur le numéro ordinal du champ.

6. Évolution du schéma et compatibilité

6.1. Niveaux de compatibilité

Chaque subject a un niveau de compatibilité. Le niveau par défaut est BACKWARD.

Niveau	Signification	Ordre de déploiement sûr
`BACKWARD`	Le nouveau schéma peut lire les anciennes données	Consumer d’abord, producer ensuite
`FORWARD`	L’ancien schéma peut lire les nouvelles données	Producer d’abord, consumer ensuite
`FULL`	Satisfait les deux directions	N’importe quel ordre
`*_TRANSITIVE`	Vérification avec toutes les anciennes versions, pas seulement la version précédente	Comme ci-dessus

Vous définissez le niveau pour chaque subject via l’API REST :

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

6.2. Un changement valide

Supposons que vous vouliez ajouter un champ currency à Order. Avec Avro, le nouveau champ doit avoir une valeur par défaut pour que le nouveau schéma atteigne FULL :

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

Avant de fusionner le code, vérifiez le nouveau schéma auprès de Registry :

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}

Si vous omettez "default": "VND", le résultat retourne false. Quand un producer essaie d’enregistrer ce schéma, il reçoit une erreur 409 Conflict.

Avec Protobuf, la règle correspondante est : ne jamais changer ou réutiliser un numéro de champ. Quand vous supprimez un champ, marquez-le comme reserved pour que personne ne réutilise ce numéro :

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

7. Écrire des tests avec un Registry simulé

Les sérialiseurs de Confluent supportent les URLs de la forme mock://<scope>. Avec cette URL, le sérialiseur utilise un Registry simulé en mémoire. Vous pouvez tester la sérialisation sans Docker :

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);
        }
    }
}

Ce test vérifie trois choses :

Le premier octet est le magic byte 0.
Les quatre octets suivants constituent un schema ID valide.
Le désérialiseur retourne la classe Order correcte, pas GenericRecord. Cela confirme que specific.avro.reader a pris effet.

Pour un test d’intégration complet, vous pouvez utiliser Testcontainers pour exécuter Kafka avec Schema Registry. Cependant, les tests avec mock:// s’exécutent rapidement et suffisent à attraper la plupart des erreurs de configuration.

Note : Les sérialiseurs utilisant le même scope mock:// partageront le même Registry simulé. Chaque classe de test devrait utiliser son propre scope pour éviter les interférences mutuelles.

8. Erreurs courantes et quand ne pas utiliser

Poison pill. Un message qui ne respecte pas le wire format fera jeter SerializationException par KafkaAvroDeserializer avant d’atteindre le listener. Le consumer peut rester bloqué à cet offset. Vous devriez envelopper le désérialiseur avec ErrorHandlingDeserializer :

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

Enregistrement automatique du schéma en production. Par défaut auto.register.schemas=true, ce qui signifie que n’importe quel producer peut enregistrer un nouveau schéma. En production, vous devriez définir auto.register.schemas: false et use.latest.version: true. Le schéma ne devrait être enregistré que via votre pipeline CI/CD, après une vérification de compatibilité.

Recevoir GenericRecord ou DynamicMessage par accident. La cause est généralement d’oublier specific.avro.reader ou specific.protobuf.value.type. La conséquence est une ClassCastException qui se produit immédiatement dans le listener.

Changer le namespace ou le nom du record dans Avro. Avec Avro, c’est un schéma complètement différent. Le specific reader ne trouvera pas la classe correspondante. Considérez le namespace comme une partie immuable du contrat.

Définir la compatibilité à NONE pour plus de commodité. Cela défait tout l’intérêt de Registry. Utilisez NONE uniquement pour les topics expérimentaux.

Note : Schema Registry est une dépendance runtime. Le sérialiseur est en cache donc Registry qui s’arrête temporairement affecte rarement les schémas connus. Mais un nouveau producer qui démarre, ou un schéma pas encore en cache, échouera immédiatement. Exécutez Registry en mode multi-instance en production.

Quand ne pas utiliser Schema Registry :

Votre système est petit, seulement une équipe possède à la fois les producers et les consumers, et il y a peu de changements. Dans ce cas, JSON avec JsonSerializer de Spring est beaucoup plus simple.
Les messages doivent être lisibles directement par des clients qui ne comprennent pas le wire format de Confluent, par exemple un outil tiers qui ne lit que le JSON pur.
Vous ne pouvez pas exploiter un service d’infrastructure supplémentaire et vous n’avez pas accès à un Registry géré.

9. Conclusion

Schema Registry transforme le format des messages Kafka en un contrat versionné avec vérification automatique. Vous avez vu comment :

Mettre en place Registry avec Docker.
Envoyer et recevoir des messages avec KafkaAvroSerializer et KafkaProtobufSerializer dans Spring Boot 3.
Comprendre le wire format composé du magic byte et du schema ID.
Définir le niveau de compatibilité et faire évoluer le schéma de manière sûre : champs nouveaux avec default pour Avro, reserved pour Protobuf.
Écrire des tests rapides avec le Registry simulé mock://.

Quand vous déployez en production, désactivez l’enregistrement automatique du schéma, intégrez la vérification de compatibilité dans votre CI, et enveloppez toujours le désérialiseur avec ErrorHandlingDeserializer. De cette façon, un changement de schéma incorrect sera arrêté dès la revue, plutôt que de casser les consumers au milieu de la nuit.