Avro et Schema Registry dans Kafka : configuration pratique

Intégration pratique d'Apache Avro et Schema Registry dans un cluster Kafka

L'intégration de Schema Registry avec un cluster Kafka permet une gestion centralisée des schémas Avro via un sujet dédié appelé _schemas. Ajoutez le service Schema Registry au docker-compose.yaml avec les serveurs bootstrap du cluster et configurez Kafka-UI pour la visualisation des schémas. Cela facilite l'évolution de la plateforme des messages JSON vers des messages typés Avro.

Configuration de Schema Registry dans Docker

Ajoutez le service schema-registry au docker-compose.yaml :

schema-registry:
  image: confluentinc/cp-schema-registry:8.1.0
  container_name: schema-registry
  hostname: schema-registry
  depends_on:
    - kafka-0
    - kafka-1
    - kafka-2
  ports:
    - "8087:8081"
  environment:
    SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka-0:9091,kafka-1:9091,kafka-2:9091
    SCHEMA_REGISTRY_HOST_NAME: schema-registry
    SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

Mettez à jour kafka-ui avec une dépendance au registre et spécifiez son adresse :

kafka-ui:
  # ... autres paramètres
  depends_on:
    - schema-registry
  environment:
    KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry:8081

Schema Registry utilise le cluster Kafka comme backend de stockage pour les schémas dans le sujet _schemas. Dans une configuration en cluster, un modèle maître-esclave est appliqué : les écritures passent uniquement par le nœud maître pour éviter les conditions de concurrence, tandis que les lectures sont gérées par les nœuds esclaves.

Mécanisme d'élection du leader dans Schema Registry

La variable SCHEMA_REGISTRY_HOST_NAME définit le nom d'hôte pour l'URL du nœud leader, qui est écrite dans le sujet _schemas. L'élection du leader utilise un group.id similaire aux consommateurs Kafka.

Sans maître-esclave, avec des requêtes parallèles à différents nœuds :

• Le client A envoie un schéma au nœud 1, voit la version 1 → crée la version 2.
• Le client B envoie simultanément au nœud 2, voit aussi la version 1 → crée une version 2 dupliquée.

Les nœuds esclaves redirigent les requêtes d'écriture vers le maître en utilisant l'URL du sujet. Cela garantit la cohérence sans synchronisation complexe.

Préparation des schémas Avro dans order-service

Créez un répertoire src/main/avro et un fichier OrderPlacedEvent.avsc :

{
  "type": "record",
  "name": "OrderPlacedEvent",
  "namespace": "io.mitochondria.order.event",
  "fields": [
    { "name": "orderId", "type": "string" },
    { "name": "email", "type": "string" },
    { "name": "productName", "type": "string" },
    { "name": "quantity", "type": "int" }
  ]
}

Le namespace détermine le package Java pour la classe générée. Supprimez tout OrderPlacedEvent existant des DTOs avant la génération. Le type quantity en int (pas une union avec null) correspond à la logique métier.

Ajoutez le dépôt Confluent au pom.xml :

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

Dépendance pour le sérialiseur :

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>8.2.0</version>
</dependency>

Dans application.properties :

spring.kafka.producer.value-serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
spring.kafka.producer.properties.schema.registry.url=http://localhost:8087

Plugin pour la génération de classes :

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals>
                <goal>schema</goal>
            </goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
                <outputDirectory>${project.basedir}/src/main/java</outputDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>

Exécutez mvn clean compile. Mettez à jour la méthode send dans kafkaTemplate :

kafkaTemplate.send("order-placed", orderPlacedEvent.getOrderId().toString(), orderPlacedEvent);

Configuration de inventory-service en tant que Producteur et Consommateur

Copiez OrderPlacedEvent.avsc dans src/main/avro dans inventory-service. Ajoutez les schémas d'événements :

InventoryRejectedEvent.avsc :

{
  "type": "record",
  "name": "InventoryRejectedEvent",
  "namespace": "io.mitochondria.inventory.event",
  "fields": [
    { "name": "orderId", "type": "string" },
    { "name": "email", "type": "string" }
  ]
}

InventoryReservedEvent.avsc :

{
  "type": "record",
  "name": "InventoryReservedEvent",
  "namespace": "io.mitochondria.inventory.event",
  "fields": [
    { "name": "orderId", "type": "string" },
    { "name": "email", "type": "string" }
  ]
}

Dans application.properties pour le producteur et le consommateur :

spring.kafka.producer.value-serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
spring.kafka.consumer.properties.spring.deserializer.value.delegate.class=io.confluent.kafka.serializers.KafkaAvroDeserializer
spring.kafka.producer.properties.schema.registry.url=http://localhost:8087
spring.kafka.consumer.properties.schema.registry.url=http://localhost:8087
spring.kafka.consumer.properties.specific.avro.reader=true

La propriété specific.avro.reader=true retourne des classes générées au lieu de GenericRecord. Supprimez les paramètres JSON et les anciennes classes d'événements, puis générez les nouvelles.

Créez des DTOs pour la logique outbox :

public record InventoryRejectedDto(String orderId, String email) {}
public record InventoryReservedDto(String orderId, String email) {}

Dans les méthodes de service, mappez les classes Avro vers les DTOs avant d'écrire dans l'outbox pour éviter de mélanger les formats.

Points clés à retenir

• Stockage des schémas : Schema Registry utilise le sujet Kafka _schemas comme backend ; le modèle maître-esclave empêche les versions dupliquées.
• Génération de code : Le plugin Avro-maven crée des classes fortement typées basées sur le namespace et s'intègre au cycle de vie de construction.
• Sérialisation : KafkaAvroSerializer enregistre et récupère automatiquement les schémas du registre via l'URL.
• Spécificités du consommateur : specific.avro.reader=true assure une désérialisation typée sans GenericRecord.
• Protection contre les conditions de concurrence : Les écritures passent uniquement par le nœud leader, avec les esclaves redirigeant les requêtes.

— Editorial Team