Avro a Schema Registry v Kafka: praxe nastavení

Praktická integrace Apache Avro a Schema Registry v Kafka clusteru

Integrace Schema Registry s Kafka clusterem umožňuje centralizovaně spravovat Avro schémata prostřednictvím speciálního topiku _schemas. Přidejte službu Schema Registry do docker-compose.yaml s bootstrap servery clusteru a nakonfigurujte Kafka-UI pro vizualizaci schémat. Tím se zajistí evoluce platformy od JSON k typizovaným Avro zprávám.

Nastavení Schema Registry v Dockeru

Přidejte službu schema-registry do 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

Aktualizujte kafka-ui s závislostí na registru a uvedením jeho adresy:

kafka-ui:
  # ... další nastavení
  depends_on:
    - schema-registry
  environment:
    KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry:8081

Schema Registry používá Kafka cluster jako úložiště schémat v topiku _schemas. V clusterové konfiguraci se používá model master-slave: zápis pouze přes leader uzel, aby se předešlo race condition, čtení – z slave uzlů.

Mechanismus výběru leadera v Schema Registry

Proměnná SCHEMA_REGISTRY_HOST_NAME určuje hostname pro URL leader uzlu, který se zapisuje do topiku _schemas. Výběr leadera používá group.id podobně jako Kafka konzumenti.

Při paralelních požadavcích na různé uzly bez master-slave:

• Klient A odešle schéma na uzel 1, vidí verzi 1 → vytvoří verzi 2.
• Klient B současně odešle na uzel 2, také vidí verzi 1 → vytvoří duplicitní verzi 2.

Slave uzly přesměrují write požadavky na master podle URL z topiku. Tím se zajistí konzistence bez složité synchronizace.

Příprava Avro schémat v order-service

Vytvořte adresář src/main/avro a soubor 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" }
  ]
}

Namespace určuje Java balíček pro vygenerovanou třídu. Odstraňte existující OrderPlacedEvent z DTO před generováním. Typ quantity jako int (ne union s null) odpovídá obchodní logice.

Přidejte Confluent repozitář do pom.xml:

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

Závislost pro serializátor:

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

V application.properties:

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

Plugin pro generování tříd:

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

Spusťte mvn clean compile. Aktualizujte odesílání v kafkaTemplate:

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

Konfigurace inventory-service jako producenta a konzumenta

Zkopírujte OrderPlacedEvent.avsc do src/main/avro inventory-service. Přidejte schémata událostí:

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

V application.properties pro producenta a konzumenta:

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

Vlastnost specific.avro.reader=true vrací vygenerované třídy místo GenericRecord. Odstraňte JSON nastavení a staré event třídy, vygenerujte nové.

Vytvořte DTO pro outbox logiku:

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

V servisní metodě mapujte Avro třídy na DTO před zápisem do outbox, aby se předešlo míchání formátů.

Co je důležité

• Ukládání schémat: Schema Registry používá Kafka topik _schemas jako backend, model master-slave zabraňuje duplicitám verzí.
• Generování kódu: Avro-maven-plugin vytváří strongly-typed třídy podle namespace, integruje se do build lifecycle.
• Serializace: KafkaAvroSerializer automaticky registruje/získává schémata z registru podle URL.
• Specifika konzumenta: specific.avro.reader=true zajišťuje typizovanou deserializaci bez GenericRecord.
• Ochrana před race condition: Zápis pouze přes leader uzel, slave přesměrovávají požadavky.

— Editorial Team