Avro und Schema Registry in Kafka: praktische Einrichtung

Praktische Integration von Apache Avro und Schema Registry in einem Kafka-Cluster

Die Integration von Schema Registry in einen Kafka-Cluster ermöglicht die zentrale Verwaltung von Avro-Schemas über ein dediziertes Thema namens _schemas. Fügen Sie den Schema-Registry-Dienst zur docker-compose.yaml mit den Bootstrap-Servern des Clusters hinzu und konfigurieren Sie Kafka-UI für die Schema-Visualisierung. Dies erleichtert die Evolution der Plattform von JSON zu typisierten Avro-Nachrichten.

Konfiguration von Schema Registry in Docker

Fügen Sie den schema-registry-Dienst zur docker-compose.yaml hinzu:

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

Aktualisieren Sie kafka-ui mit einer Abhängigkeit von der Registry und geben Sie deren Adresse an:

kafka-ui:
  # ... andere Einstellungen
  depends_on:
    - schema-registry
  environment:
    KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry:8081

Schema Registry verwendet den Kafka-Cluster als Speicher-Backend für Schemas im Thema _schemas. In einer Cluster-Konfiguration wird ein Master-Slave-Modell angewendet: Schreibvorgänge gehen nur über den Leader-Knoten, um Race Conditions zu vermeiden, während Lesevorgänge von Slave-Knoten bearbeitet werden.

Leader-Election-Mechanismus in Schema Registry

Die Variable SCHEMA_REGISTRY_HOST_NAME definiert den Hostnamen für die URL des Leader-Knotens, die in das Thema _schemas geschrieben wird. Die Leader-Election verwendet eine group.id ähnlich wie Kafka-Consumer.

Ohne Master-Slave, mit parallelen Anfragen an verschiedene Knoten:

• Client A sendet ein Schema an Knoten 1, sieht Version 1 → erstellt Version 2.
• Client B sendet gleichzeitig an Knoten 2, sieht auch Version 1 → erstellt eine doppelte Version 2.

Slave-Knoten leiten Schreibanfragen an den Master weiter, indem sie die URL aus dem Thema verwenden. Dies gewährleistet Konsistenz ohne komplexe Synchronisierung.

Vorbereitung von Avro-Schemas in order-service

Erstellen Sie ein Verzeichnis src/main/avro und eine Datei 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" }
  ]
}

Der Namespace bestimmt das Java-Paket für die generierte Klasse. Entfernen Sie vorhandene OrderPlacedEvent aus DTOs vor der Generierung. Der Typ quantity als int (keine Union mit null) entspricht der Geschäftslogik.

Fügen Sie das Confluent-Repository zu pom.xml hinzu:

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

Abhängigkeit für den Serializer:

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

In application.properties:

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

Plugin für die Klassengenerierung:

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

Führen Sie mvn clean compile aus. Aktualisieren Sie die Send-Methode in kafkaTemplate:

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

Konfiguration von inventory-service als Producer und Consumer

Kopieren Sie OrderPlacedEvent.avsc nach src/main/avro in inventory-service. Fügen Sie Event-Schemas hinzu:

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" }
  ]
}

In application.properties für Producer und Consumer:

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

Die Eigenschaft specific.avro.reader=true gibt generierte Klassen statt GenericRecord zurück. Entfernen Sie JSON-Einstellungen und alte Event-Klassen, dann generieren Sie neue.

Erstellen Sie DTOs für Outbox-Logik:

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

In Service-Methoden mappen Sie Avro-Klassen auf DTOs, bevor Sie in die Outbox schreiben, um Formate nicht zu vermischen.

Wichtige Erkenntnisse

• Schema-Speicherung: Schema Registry verwendet das Kafka-Thema _schemas als Backend; das Master-Slave-Modell verhindert doppelte Versionen.
• Code-Generierung: Das Avro-maven-Plugin erstellt stark typisierte Klassen basierend auf Namespace und integriert sie in den Build-Lebenszyklus.
• Serialisierung: KafkaAvroSerializer registriert/ruft Schemas automatisch von der Registry über URL ab.
• Consumer-Spezifika: specific.avro.reader=true gewährleistet typisierte Deserialisierung ohne GenericRecord.
• Race-Condition-Schutz: Schreibvorgänge gehen nur über den Leader-Knoten, mit Slaves, die Anfragen weiterleiten.

— Editorial Team