Avro y Schema Registry en Kafka: configuración práctica

Integración Práctica de Apache Avro y Schema Registry en un Clúster Kafka

Integrar Schema Registry con un clúster Kafka permite la gestión centralizada de esquemas Avro a través de un tema dedicado llamado _schemas. Añade el servicio Schema Registry al archivo docker-compose.yaml con los servidores bootstrap del clúster y configura Kafka-UI para la visualización de esquemas. Esto facilita la evolución de la plataforma de mensajes JSON a mensajes tipados con Avro.

Configuración de Schema Registry en Docker

Añade el servicio schema-registry a 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

Actualiza kafka-ui con una dependencia del registro y especifica su dirección:

kafka-ui:
  # ... otras configuraciones
  depends_on:
    - schema-registry
  environment:
    KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry:8081

Schema Registry utiliza el clúster Kafka como backend de almacenamiento para esquemas en el tema _schemas. En una configuración en clúster, se aplica un modelo maestro-esclavo: las escrituras solo pasan por el nodo líder para evitar condiciones de carrera, mientras que las lecturas son manejadas por los nodos esclavos.

Mecanismo de Elección de Líder en Schema Registry

La variable SCHEMA_REGISTRY_HOST_NAME define el nombre de host para la URL del nodo líder, que se escribe en el tema _schemas. La elección de líder utiliza un group.id similar a los consumidores de Kafka.

Sin maestro-esclavo, con solicitudes paralelas a diferentes nodos:

• Cliente A envía un esquema al nodo 1, ve la versión 1 → crea la versión 2.
• Cliente B simultáneamente envía al nodo 2, también ve la versión 1 → crea una versión 2 duplicada.

Los nodos esclavos redirigen las solicitudes de escritura al maestro usando la URL del tema. Esto garantiza consistencia sin sincronización compleja.

Preparación de Esquemas Avro en order-service

Crea un directorio src/main/avro y un archivo 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" }
  ]
}

El namespace determina el paquete Java para la clase generada. Elimina cualquier OrderPlacedEvent existente de los DTOs antes de la generación. El tipo quantity como int (no una unión con null) se alinea con la lógica de negocio.

Añade el repositorio de Confluent a pom.xml:

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

Dependencia para el serializador:

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

En application.properties:

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

Plugin para generación de clases:

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

Ejecuta mvn clean compile. Actualiza el método send en kafkaTemplate:

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

Configuración de inventory-service como Productor y Consumidor

Copia OrderPlacedEvent.avsc a src/main/avro en inventory-service. Añade esquemas de eventos:

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

En application.properties para productor y consumidor:

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 propiedad specific.avro.reader=true devuelve clases generadas en lugar de GenericRecord. Elimina configuraciones JSON y clases de eventos antiguas, luego genera nuevas.

Crea DTOs para lógica de outbox:

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

En métodos de servicio, mapea clases Avro a DTOs antes de escribir en outbox para evitar mezclar formatos.

Conclusiones Clave

• Almacenamiento de Esquemas: Schema Registry utiliza el tema Kafka _schemas como backend; el modelo maestro-esclavo evita versiones duplicadas.
• Generación de Código: El plugin Avro-maven crea clases fuertemente tipadas basadas en namespace y se integra en el ciclo de vida de construcción.
• Serialización: KafkaAvroSerializer registra/recupera automáticamente esquemas del registro a través de URL.
• Detalles del Consumidor: specific.avro.reader=true garantiza deserialización tipada sin GenericRecord.
• Protección contra Condiciones de Carrera: Las escrituras solo pasan por el nodo líder, con esclavos redirigiendo solicitudes.

— Editorial Team