VI
EnglishPortuguêsFrançaisDeutschTiếng ViệtEspañol简体中文한국어日本語
Trang chủ / Kiến trúc & Thiết kế / Kafka

Schema Registry và Avro/Protobuf trong Kafka với Spring Boot

Cris The Dev · 12/10/2026 · 8 phút đọc · 137

1. Tổng quan

Khi nhiều service cùng đọc và ghi một topic Kafka, rủi ro lớn nhất là định dạng message. Chỉ cần một producer đổi tên field trong JSON, các consumer phía sau có thể hỏng ngay. Confluent Schema Registry giải quyết vấn đề này. Nó lưu schema tập trung, gắn mỗi message với một schema ID và chặn những thay đổi phá vỡ tương thích.

Bài này hướng dẫn bạn dựng Schema Registry, gửi và nhận message bằng Avro và Protobuf trong Spring Boot 3.x, cấu hình compatibility, và viết test không cần Kafka thật.

2. Schema Registry hoạt động thế nào

Schema Registry là một REST service riêng, đứng cạnh Kafka cluster. Bên trong, nó lưu schema vào topic _schemas. Mỗi schema thuộc về một subject. Theo mặc định (TopicNameStrategy), subject của value trong topic orders có tên orders-value.

Khi producer gửi message, serializer làm các bước sau:

  1. Đăng ký hoặc tra cứu schema trên Registry để lấy schema ID. Kết quả được cache lại.
  2. Ghi payload theo wire format của Confluent:
| byte 0: magic byte (0) | byte 1-4: schema ID (int) | byte 5..: dữ liệu Avro/Protobuf nhị phân |

Consumer đọc 5 byte đầu, lấy schema ID, tải schema về một lần rồi cache, sau đó giải mã payload. Message không phải mang theo toàn bộ schema, nên kích thước nhỏ hơn JSON rất nhiều.

3. Dựng môi trường local

Dùng Docker Compose với Kafka chạy ở chế độ KRaft và 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

Sau khi chạy docker compose up -d, bạn kiểm tra Registry:

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

Kết quả là danh sách rỗng vì chưa có subject nào được đăng ký.

4. Avro với Spring Boot

4.1. Dependency và plugin

Các serializer của Confluent không có trên Maven Central. Bạn cần khai báo thêm repository của 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>

Plugin avro-maven-plugin sinh class Java từ file .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>

Tùy chọn stringType=String giúp field chuỗi được sinh ra kiểu java.lang.String thay vì CharSequence. Nhờ vậy, so sánh trong code và trong test đơn giản hơn.

4.2. Định nghĩa schema

Tạo file 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" }
  ]
}

Chạy mvn generate-sources. Plugin sẽ sinh class dev.nonla.order.avro.Order, kèm builder và schema nhúng sẵn trong class.

4.3. Cấu hình producer và consumer

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

Thuộc tính schema.registry.url đặt dưới spring.kafka.properties nên áp dụng cho cả producer lẫn consumer. Thuộc tính specific.avro.reader: true rất quan trọng. Nếu thiếu nó, consumer nhận về GenericRecord thay vì class Order đã sinh.

4.4. Gửi và nhận message

@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("Chuẩn bị giao đơn {} cho khách {}, giá trị {}",
                order.getId(), order.getCustomerId(), order.getAmount());
    }
}

Lần gửi đầu tiên, KafkaAvroSerializer tự đăng ký schema vào subject orders-value. Bạn có thể kiểm tra lại:

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

Lúc này message trên topic chỉ mang schema ID là 1, không mang toàn bộ schema.

5. Protobuf với Spring Boot

Các bước với Protobuf gần như giống hệt Avro, chỉ khác serializer và cách sinh code. Thêm dependency sau:

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

Định nghĩa 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;
}

Bạn sinh class Java bằng một plugin protoc cho Maven, ví dụ io.github.ascopes:protobuf-maven-plugin. Sau đó cấu hình serializer:

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

Thuộc tính specific.protobuf.value.type cho deserializer biết class đích. Nếu thiếu, bạn nhận về DynamicMessage.

Nên chọn định dạng nào?

  • Avro hợp với hệ sinh thái Kafka và data pipeline như Kafka Connect hay ksqlDB. Schema dạng JSON, quy tắc tiến hóa dựa trên tên field và giá trị mặc định.
  • Protobuf hợp khi tổ chức của bạn đã dùng gRPC. Bạn tái sử dụng được file .proto sẵn có. Quy tắc tiến hóa dựa trên số thứ tự field.

6. Tiến hóa schema và compatibility

6.1. Các mức compatibility

Mỗi subject có một mức compatibility. Mức mặc định là BACKWARD.

Mức Ý nghĩa Thứ tự deploy an toàn
BACKWARD Schema mới đọc được dữ liệu cũ Consumer trước, producer sau
FORWARD Schema cũ đọc được dữ liệu mới Producer trước, consumer sau
FULL Thỏa cả hai chiều Thứ tự nào cũng được
*_TRANSITIVE Kiểm tra với mọi version cũ, không chỉ version liền trước Như trên

Bạn đặt mức cho từng subject qua 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. Một thay đổi hợp lệ

Giả sử bạn muốn thêm field currency vào Order. Với Avro, field mới phải có giá trị mặc định thì schema mới đạt FULL:

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

Trước khi merge code, bạn nên kiểm tra schema mới với 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}

Nếu bỏ phần "default": "VND", kết quả trả về false. Khi đó producer sẽ nhận lỗi 409 Conflict lúc cố đăng ký schema.

Với Protobuf, quy tắc tương ứng là: không bao giờ đổi hoặc tái sử dụng số thứ tự field. Khi xóa một field, hãy đánh dấu reserved để không ai dùng lại số đó:

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

7. Viết test với mock Schema Registry

Serializer của Confluent hỗ trợ URL dạng mock://<scope>. Với URL này, serializer dùng một Registry giả lập trong bộ nhớ. Bạn test được việc serialize mà không cần 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);
        }
    }
}

Test này kiểm tra ba điều:

  • Byte đầu tiên là magic byte 0.
  • Bốn byte tiếp theo là một schema ID hợp lệ.
  • Deserializer trả về đúng class Order, không phải GenericRecord. Điều này xác nhận specific.avro.reader đã có hiệu lực.

Muốn có test tích hợp đầy đủ, bạn có thể dùng Testcontainers để chạy Kafka cùng Schema Registry. Tuy vậy, test với mock:// chạy nhanh và đủ để bắt phần lớn lỗi cấu hình.

Lưu ý: Các serializer dùng chung một scope mock:// sẽ chia sẻ cùng một Registry giả lập. Mỗi test class nên dùng một scope riêng để tránh ảnh hưởng lẫn nhau.

8. Lỗi hay gặp và khi nào không nên dùng

Poison pill. Một message không đúng wire format sẽ khiến KafkaAvroDeserializer ném SerializationException trước khi tới listener. Consumer có thể kẹt mãi ở offset đó. Bạn nên bọc deserializer bằng ErrorHandlingDeserializer:

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

Tự đăng ký schema trên production. Mặc định auto.register.schemas=true, tức producer nào cũng có thể đăng ký schema mới. Trên production, bạn nên đặt auto.register.schemas: false và use.latest.version: true. Schema chỉ nên được đăng ký qua pipeline CI/CD, sau bước kiểm tra compatibility.

Nhận GenericRecord hoặc DynamicMessage ngoài ý muốn. Nguyên nhân thường là quên specific.avro.reader hoặc specific.protobuf.value.type. Hậu quả là ClassCastException xảy ra ngay trong listener.

Đổi namespace hoặc tên record trong Avro. Với Avro, đây là một schema hoàn toàn khác. Specific reader sẽ không tìm thấy class tương ứng. Hãy coi namespace là phần không được đổi của hợp đồng.

Đặt compatibility NONE cho tiện. Làm vậy là mất toàn bộ lợi ích của Registry. Chỉ dùng NONE cho topic thử nghiệm.

Lưu ý: Schema Registry là một phụ thuộc runtime. Serializer có cache nên Registry chết tạm thời ít khi ảnh hưởng tới schema đã biết. Nhưng một producer mới khởi động, hoặc một schema chưa có trong cache, sẽ lỗi ngay. Hãy chạy Registry ở chế độ nhiều instance trên production.

Khi nào không nên dùng Schema Registry:

  • Hệ thống nhỏ, chỉ một team sở hữu cả producer lẫn consumer, ít thay đổi. Khi đó JSON kèm JsonSerializer của Spring đơn giản hơn nhiều.
  • Message cần được đọc trực tiếp bởi các client không hiểu wire format của Confluent, ví dụ một công cụ bên thứ ba chỉ đọc JSON thuần.
  • Bạn không thể vận hành thêm một service hạ tầng và cũng không có Registry dạng managed.

9. Kết luận

Schema Registry biến định dạng message Kafka thành một hợp đồng có version và có kiểm tra tự động. Bạn đã thấy cách:

  • Dựng Registry bằng Docker.
  • Gửi và nhận message bằng KafkaAvroSerializer và KafkaProtobufSerializer trong Spring Boot 3.
  • Hiểu wire format gồm magic byte và schema ID.
  • Đặt mức compatibility và tiến hóa schema an toàn: field mới có default với Avro, reserved với Protobuf.
  • Viết test nhanh bằng mock:// Registry.

Khi áp dụng vào production, hãy tắt tự đăng ký schema, đưa bước kiểm tra compatibility vào CI, và luôn bọc deserializer bằng ErrorHandlingDeserializer. Như vậy, một thay đổi sai schema sẽ bị chặn ngay từ lúc review, thay vì làm hỏng consumer lúc nửa đêm.

kafkaschema-registryavroprotobufspring-boot
Ứng dụng của mìnhHoanTien.proMua Shopee, TikTok, Lazada được hoàn tiềniOS · Android · Web

Bài cùng chủ đề

Kafka · 12/10/2026 · 8 phút đọc · 312

Change Data Capture với Debezium và Kafka

Dùng Debezium đọc WAL của PostgreSQL, đẩy mọi thay đổi dữ liệu vào Kafka và tiêu thụ chúng bằng Spring Boot, không cần sửa code ghi dữ liệu.