Avro i Schema Registry w Kafka: praktyka konfiguracji

Praktyczna integracja Apache Avro i Schema Registry w klastrze Kafka

Integracja Schema Registry z klastrem Kafka umożliwia scentralizowane zarządzanie schematami Avro poprzez specjalny temat _schemas. Dodaj usługę Schema Registry do docker-compose.yaml z serwerami bootstrap klastra i skonfiguruj Kafka-UI do wizualizacji schematów. Zapewnia to ewolucję platformy z JSON na typowane wiadomości Avro.

Konfiguracja Schema Registry w Docker

Dodaj usługę 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

Zaktualizuj kafka-ui z zależnością od rejestru i podaniem jego adresu:

kafka-ui:
  # ... inne ustawienia
  depends_on:
    - schema-registry
  environment:
    KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry:8081

Schema Registry używa klastra Kafka jako magazynu schematów w temacie _schemas. W konfiguracji klastrowej stosowany jest model master-slave: zapis tylko przez węzeł leader, aby uniknąć wyścigu, odczyt — z węzłów slave.

Mechanizm wyboru leadera w Schema Registry

Zmienna SCHEMA_REGISTRY_HOST_NAME określa hostname dla URL węzła leader, zapisywanego w temacie _schemas. Wybór leadera wykorzystuje group.id podobnie jak konsumenci Kafka.

Przy równoległych żądaniach do różnych węzłów bez master-slave:

• Klient A wysyła schemat do węzła 1, widzi wersję 1 → tworzy wersję 2.
• Klient B jednocześnie wysyła do węzła 2, również widzi wersję 1 → tworzy duplikującą wersję 2.

Węzły slave przekierowują żądania zapisu do mastera przez URL z tematu. Zapewnia to spójność bez złożonej synchronizacji.

Przygotowanie schematów Avro w order-service

Utwórz katalog src/main/avro i plik 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 definiuje pakiet Java dla wygenerowanej klasy. Usuń istniejący OrderPlacedEvent z DTO przed generacją. Typ quantity jako int (nie union z null) odpowiada logice biznesowej.

Dodaj repozytorium Confluent w pom.xml:

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

Zależność dla serializatora:

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

W application.properties:

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

Wtyczka do generowania klas:

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

Wykonaj mvn clean compile. Zaktualizuj wysyłanie w kafkaTemplate:

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

Konfiguracja inventory-service jako producenta i konsumenta

Skopiuj OrderPlacedEvent.avsc do src/main/avro inventory-service. Dodaj schematy zdarzeń:

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

W application.properties dla producenta i konsumenta:

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

Właściwość specific.avro.reader=true zwraca wygenerowane klasy zamiast GenericRecord. Usuń ustawienia JSON i stare klasy eventów, wygeneruj nowe.

Utwórz DTO dla logiki outbox:

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

W metodzie serwisowej mapuj klasy Avro na DTO przed zapisem w outbox, unikając mieszania formatów.

Co jest ważne

• Przechowywanie schematów: Schema Registry używa tematu Kafka _schemas jako backendu, model master-slave zapobiega duplikatom wersji.
• Generowanie kodu: Avro-maven-plugin tworzy silnie typowane klasy według namespace, integruje się w cykl budowania.
• Serializacja: KafkaAvroSerializer automatycznie rejestruje/pobiera schematy z rejestru przez URL.
• Specyfika konsumenta: specific.avro.reader=true zapewnia typowaną deserializację bez GenericRecord.
• Ochrona przed wyścigiem: Zapis tylko przez węzeł leader, slave przekierowują żądania.

— Editorial Team