Avro and Schema Registry in Kafka: practical setup

Practical Integration of Apache Avro and Schema Registry in a Kafka Cluster

Integrating Schema Registry with a Kafka cluster allows for centralized management of Avro schemas through a dedicated topic called _schemas. Add the Schema Registry service to docker-compose.yaml with the cluster's bootstrap servers and configure Kafka-UI for schema visualization. This facilitates the platform's evolution from JSON to typed Avro messages.

Configuring Schema Registry in Docker

Add the schema-registry service to 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

Update kafka-ui with a dependency on the registry and specify its address:

kafka-ui:
  # ... other settings
  depends_on:
    - schema-registry
  environment:
    KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry:8081

Schema Registry uses the Kafka cluster as a storage backend for schemas in the _schemas topic. In a clustered configuration, a master-slave model is applied: writes only go through the leader node to avoid race conditions, while reads are handled by slave nodes.

Leader Election Mechanism in Schema Registry

The SCHEMA_REGISTRY_HOST_NAME variable defines the hostname for the leader node's URL, which is written to the _schemas topic. Leader election uses a group.id similar to Kafka consumers.

Without master-slave, with parallel requests to different nodes:

• Client A sends a schema to node 1, sees version 1 → creates version 2.
• Client B simultaneously sends to node 2, also sees version 1 → creates a duplicate version 2.

Slave nodes redirect write requests to the master using the URL from the topic. This ensures consistency without complex synchronization.

Preparing Avro Schemas in order-service

Create a directory src/main/avro and a file 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" }
  ]
}

The namespace determines the Java package for the generated class. Remove any existing OrderPlacedEvent from DTOs before generation. The quantity type as int (not a union with null) aligns with business logic.

Add the Confluent repository to pom.xml:

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

Dependency for the 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 for class generation:

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

Run mvn clean compile. Update the send method in kafkaTemplate:

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

Configuring inventory-service as Producer and Consumer

Copy OrderPlacedEvent.avsc to src/main/avro in inventory-service. Add event schemas:

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 for producer and 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

The property specific.avro.reader=true returns generated classes instead of GenericRecord. Remove JSON settings and old event classes, then generate new ones.

Create DTOs for outbox logic:

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

In service methods, map Avro classes to DTOs before writing to outbox to avoid mixing formats.

Key Takeaways

• Schema Storage: Schema Registry uses the Kafka topic _schemas as a backend; the master-slave model prevents duplicate versions.
• Code Generation: The Avro-maven-plugin creates strongly-typed classes based on namespace and integrates into the build lifecycle.
• Serialization: KafkaAvroSerializer automatically registers/retrieves schemas from the registry via URL.
• Consumer Specifics: specific.avro.reader=true ensures typed deserialization without GenericRecord.
• Race Condition Protection: Writes only go through the leader node, with slaves redirecting requests.

— Editorial Team