PT
EnglishPortuguêsFrançaisDeutschTiếng ViệtEspañol简体中文한국어日本語
Início / Arquitetura e design / Kafka

Schema Registry e Avro/Protobuf em Kafka com Spring Boot

Cris The Dev · 12/10/2026 · 8 min de leitura · 137

1. Visão geral

Quando múltiplos serviços leem e escrevem o mesmo tópico Kafka, o maior risco é o formato da mensagem. É só um produtor mudar o nome de um field no JSON para os consumers downstream quebrarem na hora. O Confluent Schema Registry resolve isso. Ele armazena schemas centralizados, vincula cada mensagem a um schema ID e impede mudanças que quebram compatibilidade.

Este post mostra como montar um Schema Registry, enviar e receber mensagens com Avro e Protobuf no Spring Boot 3.x, configurar compatibilidade e escrever testes sem precisar de um Kafka real.

2. Como funciona o Schema Registry

O Schema Registry é um serviço REST separado, rodando ao lado do cluster Kafka. Internamente, ele armazena os schemas em um tópico especial chamado _schemas. Cada schema pertence a um subject. Por padrão (TopicNameStrategy), o subject para o value do tópico orders é nomeado orders-value.

Quando um produtor envia uma mensagem, o serializador executa os seguintes passos:

  1. Registra ou busca o schema no Registry para obter o schema ID. O resultado é armazenado em cache.
  2. Escreve o payload no wire format do Confluent:
| byte 0: magic byte (0) | bytes 1-4: schema ID (int) | bytes 5+: dados Avro/Protobuf em binário |

O consumidor lê os primeiros 5 bytes, extrai o schema ID, carrega o schema uma vez e faz cache dele, depois decodifica o payload. A mensagem não precisa carregar o schema inteiro, então é muito menor que JSON.

3. Configurar o ambiente local

Use Docker Compose com Kafka rodando em modo KRaft e Schema Registry:

services:
  kafka:
    image: confluentinc/cp-kafka:7.7.1
    ports: ["9092:9092"]
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:29093
      KAFKA_LISTENERS: PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,EXTERNAL://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

  schema-registry:
    image: confluentinc/cp-schema-registry:7.7.1
    depends_on: [kafka]
    ports: ["8081:8081"]
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKA_STORE_BOOTSTRAP_SERVERS: kafka:29092
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

Depois de executar docker compose up -d, você verifica o Registry:

curl -s http://localhost:8081/subjects
[]

O resultado é uma lista vazia porque nenhum subject foi registrado ainda.

4. Avro com Spring Boot

4.1. Dependências e plugin

Os serializadores do Confluent não estão no Maven Central. Você precisa declarar o repositório do Confluent:

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

<dependencies>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-avro-serializer</artifactId>
        <version>7.7.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>1.12.0</version>
    </dependency>
</dependencies>

O plugin avro-maven-plugin gera classes Java a partir de arquivos .avsc:

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.12.0</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals><goal>schema</goal></goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
                <outputDirectory>${project.build.directory}/generated-sources/avro</outputDirectory>
                <stringType>String</stringType>
            </configuration>
        </execution>
    </executions>
</plugin>

A opção stringType=String faz com que campos de string sejam gerados como java.lang.String em vez de CharSequence. Assim, comparações no código e testes ficam mais simples.

4.2. Definindo o schema

Crie o arquivo src/main/avro/Order.avsc:

{
  "type": "record",
  "name": "Order",
  "namespace": "dev.nonla.order.avro",
  "fields": [
    { "name": "id", "type": "string" },
    { "name": "customerId", "type": "string" },
    { "name": "amount", "type": "double" }
  ]
}

Execute mvn generate-sources. O plugin vai gerar a classe dev.nonla.order.avro.Order, com builder e schema embutido na classe.

4.3. Configurando produtor e consumidor

spring:
  kafka:
    bootstrap-servers: localhost:9092
    properties:
      schema.registry.url: http://localhost:8081
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
    consumer:
      group-id: shipping-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      properties:
        specific.avro.reader: true

A propriedade schema.registry.url sob spring.kafka.properties se aplica tanto ao produtor quanto ao consumidor. A propriedade specific.avro.reader: true é crucial. Sem ela, o consumidor recebe GenericRecord em vez da classe Order gerada.

4.4. Enviando e recebendo mensagens

@Service
public class OrderPublisher {

    private final KafkaTemplate<String, Order> kafkaTemplate;

    public OrderPublisher(KafkaTemplate<String, Order> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publish(String orderId, String customerId, double amount) {
        Order order = Order.newBuilder()
                .setId(orderId)
                .setCustomerId(customerId)
                .setAmount(amount)
                .build();
        kafkaTemplate.send("orders", order.getId(), order);
    }
}
@Component
public class ShippingListener {

    private static final Logger log = LoggerFactory.getLogger(ShippingListener.class);

    @KafkaListener(topics = "orders")
    public void onOrder(Order order) {
        log.info("Preparando entrega do pedido {} para cliente {}, valor {}",
                order.getId(), order.getCustomerId(), order.getAmount());
    }
}

Na primeira vez que você envia, KafkaAvroSerializer registra automaticamente o schema para o subject orders-value. Você pode verificar:

curl -s http://localhost:8081/subjects/orders-value/versions/latest
{"subject":"orders-value","version":1,"id":1,"schema":"{\"type\":\"record\",\"name\":\"Order\",...}"}

Neste ponto, as mensagens no tópico carregam apenas o schema ID 1, não o schema completo.

5. Protobuf com Spring Boot

Os passos com Protobuf são praticamente idênticos aos de Avro, diferindo apenas no serializador e na geração de código. Adicione a seguinte dependência:

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-protobuf-serializer</artifactId>
    <version>7.7.1</version>
</dependency>

Defina src/main/proto/order_event.proto:

syntax = "proto3";

package dev.nonla.order;

option java_package = "dev.nonla.order.proto";
option java_multiple_files = true;

message OrderEvent {
  string id = 1;
  string customer_id = 2;
  double amount = 3;
}

Você gera classes Java usando um plugin protoc para Maven, por exemplo io.github.ascopes:protobuf-maven-plugin. Depois configure o serializador:

spring:
  kafka:
    producer:
      value-serializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer
    consumer:
      value-deserializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer
      properties:
        specific.protobuf.value.type: dev.nonla.order.proto.OrderEvent

A propriedade specific.protobuf.value.type informa ao deserializador a classe de destino. Sem ela, você recebe DynamicMessage.

Qual formato escolher?

  • Avro funciona bem com o ecossistema Kafka e pipelines de dados como Kafka Connect ou ksqlDB. Schemas em JSON, regras de evolução baseadas em nomes de field e valores padrão.
  • Protobuf é adequado quando sua organização já usa gRPC. Você pode reutilizar arquivos .proto existentes. Regras de evolução baseadas no número ordinal do field.

6. Evolução de schema e compatibilidade

6.1. Níveis de compatibilidade

Cada subject tem um nível de compatibilidade. O nível padrão é BACKWARD.

Nível Significado Ordem de deploy segura
BACKWARD Schema novo lê dados antigos Consumidor primeiro, produtor depois
FORWARD Schema antigo lê dados novos Produtor primeiro, consumidor depois
FULL Satisfaz ambas as direções Qualquer ordem
*_TRANSITIVE Verifica com todas as versões antigas, não só a anterior Como acima

Você define o nível por subject via REST API:

curl -s -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility": "FULL_TRANSITIVE"}' \
  http://localhost:8081/config/orders-value

6.2. Uma mudança válida

Suponha que você queira adicionar o field currency ao Order. Com Avro, o novo field deve ter um valor padrão para que o schema novo alcance FULL:

{ "name": "currency", "type": "string", "default": "VND" }

Antes de fazer merge, você deve validar o novo schema no Registry:

curl -s -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"dev.nonla.order.avro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"customerId\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"currency\",\"type\":\"string\",\"default\":\"VND\"}]}"}' \
  http://localhost:8081/compatibility/subjects/orders-value/versions/latest
{"is_compatible":true}

Se você omitir "default": "VND", o resultado será false. Nesse caso, o produtor receberá um erro 409 Conflict ao tentar registrar o schema.

Com Protobuf, a regra correspondente é: nunca mude ou reutilize o número ordinal de um field. Quando remover um field, marque como reserved para que ninguém reutilize esse número:

message OrderEvent {
  reserved 2;
  reserved "customer_id";
  string id = 1;
  double amount = 3;
  string currency = 4;
}

7. Testando com mock Schema Registry

O serializador do Confluent suporta URLs no formato mock://<scope>. Com essa URL, o serializador usa um Registry simulado na memória. Você pode testar serialização sem precisar de Docker:

import static org.assertj.core.api.Assertions.assertThat;

class OrderAvroSerdeTest {

    private static final Map<String, Object> CONFIG = Map.of(
            "schema.registry.url", "mock://order-serde-test",
            "specific.avro.reader", true);

    @Test
    void shouldSerializeWithWireFormatAndDeserializeBack() {
        try (var serializer = new KafkaAvroSerializer();
             var deserializer = new KafkaAvroDeserializer()) {
            serializer.configure(CONFIG, false);
            deserializer.configure(CONFIG, false);

            Order order = Order.newBuilder()
                    .setId("ORD-1001")
                    .setCustomerId("CUS-42")
                    .setAmount(250_000.0)
                    .build();

            byte[] bytes = serializer.serialize("orders", order);

            assertThat(bytes[0]).isZero();                       // magic byte
            assertThat(ByteBuffer.wrap(bytes, 1, 4).getInt()).isPositive(); // schema ID

            Object result = deserializer.deserialize("orders", bytes);
            assertThat(result).isInstanceOf(Order.class).isEqualTo(order);
        }
    }
}

Este teste verifica três coisas:

  • O primeiro byte é o magic byte 0.
  • Os próximos quatro bytes formam um schema ID válido.
  • O deserializador retorna a classe correta Order, não GenericRecord. Isso confirma que specific.avro.reader está ativo.

Para testes de integração completos, você pode usar Testcontainers para executar Kafka com Schema Registry. Porém, testes com mock:// são rápidos e suficientes para capturar a maioria dos erros de configuração.

Nota: Serializadores que usam o mesmo scope mock:// compartilham um Registry simulado. Cada classe de teste deve usar um scope diferente para evitar interferência.

8. Erros comuns e quando não usar

Mensagem envenenada. Uma mensagem com wire format incorreto faz KafkaAvroDeserializer lançar SerializationException antes de chegar ao listener. O consumidor pode ficar preso nesse offset para sempre. Você deve envolver o deserializador com ErrorHandlingDeserializer:

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: io.confluent.kafka.serializers.KafkaAvroDeserializer

Auto-registro de schema em produção. Por padrão auto.register.schemas=true, significa que qualquer produtor pode registrar um novo schema. Em produção, você deve definir auto.register.schemas: false e use.latest.version: true. Schemas devem ser registrados apenas através do pipeline CI/CD, após validação de compatibilidade.

Receber GenericRecord ou DynamicMessage inesperadamente. A causa geralmente é esquecer specific.avro.reader ou specific.protobuf.value.type. A consequência é ClassCastException lançado diretamente no listener.

Mudando namespace ou nome do record em Avro. Com Avro, isso é um schema completamente diferente. O specific reader não encontrará a classe correspondente. Considere o namespace como a parte imutável do contrato.

Definir compatibilidade como NONE por conveniência. Fazer isso anula todos os benefícios do Registry. Use NONE apenas para tópicos de experimentação.

Nota: Schema Registry é uma dependência em runtime. O serializador faz cache, então Registry cair temporariamente raramente afeta schemas já conhecidos. Mas um novo produtor iniciando ou um schema não no cache falhará imediatamente. Execute o Registry em modo multi-instância em produção.

Quando não usar Schema Registry:

  • Sistema pequeno, apenas um time possui tanto o produtor quanto o consumidor, poucas mudanças. Nesse caso, JSON com JsonSerializer do Spring é muito mais simples.
  • Mensagens precisam ser lidas diretamente por clientes que não entendem o wire format do Confluent, por exemplo uma ferramenta de terceiros que só lê JSON puro.
  • Você não consegue operar um serviço de infraestrutura adicional e também não tem um Registry gerenciado disponível.

9. Conclusão

Schema Registry transforma o formato de mensagem Kafka em um contrato com versionamento e validação automática. Você viu como:

  • Montar um Registry com Docker.
  • Enviar e receber mensagens com KafkaAvroSerializer e KafkaProtobufSerializer no Spring Boot 3.
  • Entender o wire format composto de magic byte e schema ID.
  • Definir níveis de compatibilidade e evoluir schemas com segurança: novo field com padrão em Avro, reserved em Protobuf.

Ao aplicar em produção, desative o auto-registro de schemas, coloque a validação de compatibilidade no CI e sempre envolva o deserializador com ErrorHandlingDeserializer. Assim, uma mudança de schema inválida será bloqueada na revisão, em vez de quebrar consumers no meio da noite.

kafkaschema-registryavroprotobufspring-boot
App do autorHoanTien.proCashback on Shopee, TikTok Shop and Lazada (Vietnam)iOS · Android · Web

Mais sobre este tema

Kafka · 12/10/2026 · 8 min de leitura · 312

Change Data Capture com Debezium e Kafka

Use Debezium para ler o WAL do PostgreSQL, enviar todas as mudanças de dados para Kafka e consumi-las com Spring Boot, sem precisar modificar o código de escrita de dados.