Schema Registry와 Avro/Protobuf in Kafka with Spring Boot

1. 개요

여러 서비스가 하나의 Kafka 토픽에서 읽고 쓸 때, 가장 큰 위험은 메시지 형식입니다. 한 producer가 JSON의 필드명을 바꾸면 뒤의 consumer들이 즉시 깨질 수 있습니다. Confluent Schema Registry가 이 문제를 해결합니다. 이것은 스키마를 중앙에서 관리하고, 각 메시지에 스키마 ID를 붙이며, 호환성을 깨뜨리는 변경을 방지합니다.

이 글은 Schema Registry를 구축하고, Spring Boot 3.x에서 Avro와 Protobuf로 메시지를 주고받으며, 호환성을 설정하고, 실제 Kafka 없이 테스트를 작성하는 방법을 안내합니다.

2. Schema Registry는 어떻게 동작하나요

Schema Registry는 Kafka 클러스터 옆에 서 있는 별도의 REST 서비스입니다. 내부적으로 스키마를 _schemas 토픽에 저장합니다. 각 스키마는 하나의 subject에 속합니다. 기본적으로(TopicNameStrategy), orders 토픽의 값(value)의 subject 이름은 orders-value입니다.

Producer가 메시지를 보낼 때, serializer는 다음 단계를 거칩니다:

1. Registry에 스키마를 등록하거나 조회해서 스키마 ID를 얻습니다. 결과는 캐시됩니다.
2. Confluent의 wire format에 따라 페이로드를 씁니다:

| byte 0: magic byte (0) | byte 1-4: schema ID (int) | byte 5..: Avro/Protobuf 바이너리 데이터 |

Consumer는 처음 5바이트를 읽어서 스키마 ID를 얻고, 스키마를 한 번 로드한 뒤 캐시한 다음, 페이로드를 디코딩합니다. 메시지가 전체 스키마를 가지고 있지 않아도 되므로, JSON보다 크기가 훨씬 작습니다.

3. 로컬 환경 구축

KRaft 모드에서 실행되는 Kafka와 Schema Registry를 Docker Compose로 실행합니다:

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

docker compose up -d를 실행한 후, Registry를 확인합니다:

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

[]

결과는 빈 리스트입니다. 아직 등록된 subject가 없기 때문입니다.

4. Spring Boot에서 Avro 사용하기

4.1. Dependency와 plugin

Confluent의 serializer는 Maven Central에 없습니다. Confluent repository를 선언해야 합니다:

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

avro-maven-plugin은 .avsc 파일로부터 Java 클래스를 생성합니다:

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

stringType=String 옵션은 문자열 필드를 CharSequence 대신 java.lang.String으로 생성하게 하므로, 코드와 테스트에서 비교가 더 간단합니다.

4.2. 스키마 정의

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

mvn generate-sources를 실행합니다. Plugin이 dev.nonla.order.avro.Order 클래스를 생성합니다. 이 클래스는 builder와 스키마가 내장되어 있습니다.

4.3. Producer와 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

schema.registry.url 속성은 spring.kafka.properties 아래에 놓아서 producer와 consumer 모두에게 적용되어야 합니다. specific.avro.reader: true 속성이 매우 중요합니다. 이것이 없으면, consumer는 생성된 Order 클래스 대신 GenericRecord를 받습니다.

4.4. 메시지 주고받기

@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("주문 {}을(를) 고객 {}에게 배송 준비 중. 금액: {}",
                order.getId(), order.getCustomerId(), order.getAmount());
    }
}

첫 전송 때, KafkaAvroSerializer는 자동으로 스키마를 subject orders-value에 등록합니다. 다시 확인할 수 있습니다:

curl -s http://localhost:8081/subjects/orders-value/versions/latest

{"subject":"orders-value","version":1,"id":1,"schema":"{\"type\":\"record\",\"name\":\"Order\",...}"}

이제 토픽의 메시지는 스키마 ID 1만 들고 있고, 전체 스키마는 들고 있지 않습니다.

5. Spring Boot에서 Protobuf 사용하기

Protobuf 단계는 Avro와 거의 같습니다. serializer와 코드 생성 방식만 다릅니다. 다음 dependency를 추가합니다:

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

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

Maven용 protoc plugin(예: io.github.ascopes:protobuf-maven-plugin)으로 Java 클래스를 생성합니다. 그 다음 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

specific.protobuf.value.type 속성은 deserializer에 대상 클래스를 알려줍니다. 이것이 없으면 DynamicMessage를 받습니다.

어느 형식을 선택할까요?

• Avro는 Kafka 생태계와 Kafka Connect, ksqlDB 같은 data pipeline과 잘 맞습니다. 스키마는 JSON이고, 진화 규칙은 필드명과 기본값을 기반으로 합니다.
• Protobuf는 조직이 이미 gRPC를 사용할 때 적합합니다. 기존 .proto 파일을 재사용할 수 있습니다. 진화 규칙은 필드 번호를 기반으로 합니다.

6. 스키마 진화와 호환성

6.1. 호환성 레벨

각 subject는 하나의 호환성 레벨을 가집니다. 기본값은 BACKWARD입니다.

레벨	의미	배포 순서
BACKWARD	새 스키마는 기존 데이터를 읽을 수 있음	Consumer 먼저, Producer 나중
FORWARD	기존 스키마는 새 데이터를 읽을 수 있음	Producer 먼저, Consumer 나중
FULL	양쪽 다 만족	어느 순서든
*_TRANSITIVE	모든 기존 버전과 호환성을 확인함 (바로 앞 버전만 아님)	위와 같음

REST API로 각 subject의 레벨을 설정합니다:

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

6.2. 유효한 변경 사항

Order에 currency 필드를 추가한다고 하겠습니다. Avro에서 새 필드는 새 스키마가 FULL을 만족하려면 반드시 기본값을 가져야 합니다:

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

코드를 merge하기 전에, 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}

"default": "VND" 부분을 빼면 결과는 false입니다. 이 경우 producer는 스키마를 등록하려 할 때 409 Conflict 오류를 받습니다.

Protobuf에서 해당하는 규칙은: 필드 번호를 절대 바꾸거나 재사용하지 마세요. 필드를 제거할 때는, 그 번호를 다시 쓰지 않도록 reserved로 표시하세요:

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

7. Mock Schema Registry로 테스트 작성하기

Confluent serializer는 mock://<scope> 형태의 URL을 지원합니다. 이 URL을 쓰면 serializer가 메모리 내 가짜 Registry를 사용합니다. Docker 없이 serialize를 테스트할 수 있습니다:

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);
        }
    }
}

이 테스트는 세 가지를 확인합니다:

• 첫 번째 바이트는 magic byte 0입니다.
• 다음 4바이트는 유효한 스키마 ID입니다.
• Deserializer가 GenericRecord 대신 정확히 Order 클래스를 반환합니다. 이는 specific.avro.reader가 효과가 있음을 확인합니다.

완전한 통합 테스트를 원한다면, Testcontainers를 사용해 Kafka와 Schema Registry를 함께 실행할 수 있습니다. 다만 mock://로 하는 테스트가 훨씬 빠르고 대부분의 설정 오류를 잡기에 충분합니다.

주의: mock://을 같은 scope로 사용하는 serializer들은 같은 가짜 Registry를 공유합니다. 각 테스트 클래스는 서로 다른 scope를 써야 서로 영향을 주지 않습니다.

8. 흔한 오류와 사용하지 말아야 할 때

Poison pill. 올바른 wire format이 아닌 메시지는 KafkaAvroDeserializer가 listener에 도달하기 전에 SerializationException을 던집니다. Consumer가 그 offset에 영구히 걸릴 수 있습니다. ErrorHandlingDeserializer로 deserializer를 감싸야 합니다:

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

프로덕션에서 자동 스키마 등록. 기본값은 auto.register.schemas=true로, 모든 producer가 새 스키마를 등록할 수 있습니다. 프로덕션에서는 auto.register.schemas: false와 use.latest.version: true를 설정하세요. 스키마는 호환성 검사를 거친 CI/CD 파이프라인을 통해서만 등록되어야 합니다.

의도하지 않게 GenericRecord 또는 DynamicMessage 받기. 보통 specific.avro.reader나 specific.protobuf.value.type을 빠뜨린 것이 원인입니다. 결과는 listener 안에서 ClassCastException이 발생합니다.

Avro에서 namespace 또는 record 이름 바꾸기. Avro에서 이것은 완전히 다른 스키마입니다. Specific reader는 해당하는 클래스를 찾지 못합니다. namespace는 계약의 불변 부분이라고 생각하세요.

편하려고 호환성을 NONE으로 설정하기. 그렇게 하면 Registry의 모든 이점이 없어집니다. NONE은 테스트 토픽에만 사용하세요.

주의: Schema Registry는 runtime 의존성입니다. Serializer는 캐시를 가지므로 Registry가 일시적으로 다운되어도 알려진 스키마는 잘 작동합니다. 하지만 새로 시작한 producer나 캐시에 없는 스키마는 즉시 오류가 납니다. 프로덕션에서는 Registry를 여러 인스턴스로 실행하세요.

사용하지 말아야 할 때:

• 시스템이 작고, 한 팀만 producer와 consumer를 소유하며, 거의 변경이 없을 때. 이 경우 Spring의 JsonSerializer와 함께 JSON을 쓰는 것이 훨씬 간단합니다.
• 메시지를 Confluent의 wire format을 이해하지 못하는 외부 클라이언트가 직접 읽어야 할 때. 예: 서드파티 도구가 순수 JSON만 읽을 때.
• 추가 인프라 서비스를 운영할 수 없고, managed Registry도 없을 때.

9. 결론

Schema Registry는 Kafka 메시지 형식을 버전이 있고 자동으로 확인되는 계약으로 바꿉니다. 여러분은 이제 다음을 알았습니다:

• Docker로 Registry를 구축하는 방법.
• Spring Boot 3에서 KafkaAvroSerializer와 KafkaProtobufSerializer로 메시지를 주고받는 방법.
• Magic byte와 스키마 ID로 이루어진 wire format 이해.
• 호환성 레벨을 설정하고 안전하게 스키마를 진화시키는 방법: Avro는 새 필드에 기본값, Protobuf는 reserved.
• mock:// Registry로 빠른 테스트를 작성하는 방법.

프로덕션에 적용할 때는 자동 스키마 등록을 끄고, CI에 호환성 검사 단계를 넣고, deserializer를 항상 ErrorHandlingDeserializer로 감싸세요. 그러면 스키마 오류가 review 단계에서 잡혀서 밤중에 consumer를 깨뜨리지 않습니다.