Kafka의 Avro와 Schema Registry: 실전 설정

Kafka 클러스터에서 Apache Avro와 Schema Registry의 실용적 통합

Schema Registry를 Kafka 클러스터와 통합하면 _schemas라는 전용 토픽을 통해 Avro 스키마를 중앙 집중식으로 관리할 수 있습니다. docker-compose.yaml에 Schema Registry 서비스를 추가하고 클러스터의 부트스트랩 서버를 구성하며, 스키마 시각화를 위해 Kafka-UI를 설정합니다. 이를 통해 플랫폼이 JSON에서 타입이 지정된 Avro 메시지로 진화하는 데 도움이 됩니다.

Docker에서 Schema Registry 구성

docker-compose.yaml에 schema-registry 서비스를 추가합니다:

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

kafka-ui를 레지스트리에 종속되도록 업데이트하고 주소를 지정합니다:

kafka-ui:
  # ... 기타 설정
  depends_on:
    - schema-registry
  environment:
    KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry:8081

Schema Registry는 Kafka 클러스터를 _schemas 토픽에 스키마를 저장하는 백엔드로 사용합니다. 클러스터 구성에서는 마스터-슬레이브 모델이 적용됩니다: 경쟁 조건을 피하기 위해 쓰기는 리더 노드에서만 처리되고, 읽기는 슬레이브 노드에서 처리됩니다.

Schema Registry의 리더 선출 메커니즘

SCHEMA_REGISTRY_HOST_NAME 변수는 리더 노드의 URL에 대한 호스트명을 정의하며, 이는 _schemas 토픽에 기록됩니다. 리더 선출은 Kafka 컨슈머와 유사한 group.id를 사용합니다.

마스터-슬레이브 없이, 다른 노드에 병렬 요청이 있을 때:

• 클라이언트 A가 노드 1에 스키마를 보내고, 버전 1을 확인 → 버전 2를 생성합니다.
• 클라이언트 B가 동시에 노드 2에 보내고, 버전 1을 확인 → 중복된 버전 2를 생성합니다.

슬레이브 노드는 토픽의 URL을 사용하여 쓰기 요청을 마스터로 리디렉션합니다. 이로써 복잡한 동기화 없이 일관성을 보장합니다.

order-service에서 Avro 스키마 준비

src/main/avro 디렉토리와 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" }
  ]
}

네임스페이스는 생성된 클래스의 Java 패키지를 결정합니다. 생성 전에 DTO에서 기존 OrderPlacedEvent를 제거합니다. quantity 타입을 int로 설정하는 것은(널과의 유니온이 아님) 비즈니스 로직과 일치합니다.

pom.xml에 Confluent 저장소를 추가합니다:

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

직렬화를 위한 의존성:

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

application.properties에서:

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

클래스 생성을 위한 플러그인:

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

mvn clean compile을 실행합니다. kafkaTemplate의 send 메서드를 업데이트합니다:

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

inventory-service를 프로듀서와 컨슈머로 구성

OrderPlacedEvent.avsc를 inventory-service의 src/main/avro에 복사합니다. 이벤트 스키마를 추가합니다:

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

프로듀서와 컨슈머를 위한 application.properties에서:

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

specific.avro.reader=true 속성은 GenericRecord 대신 생성된 클래스를 반환합니다. JSON 설정과 기존 이벤트 클래스를 제거한 후 새로운 클래스를 생성합니다.

아웃박스 로직을 위한 DTO를 생성합니다:

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

서비스 메서드에서, 형식 혼합을 피하기 위해 아웃박스에 쓰기 전에 Avro 클래스를 DTO로 매핑합니다.

핵심 요약

• 스키마 저장: Schema Registry는 Kafka 토픽 _schemas를 백엔드로 사용하며, 마스터-슬레이브 모델이 중복 버전을 방지합니다.
• 코드 생성: Avro-maven-plugin은 네임스페이스를 기반으로 강력한 타입의 클래스를 생성하고 빌드 라이프사이클에 통합합니다.
• 직렬화: KafkaAvroSerializer는 URL을 통해 레지스트리에서 스키마를 자동으로 등록/검색합니다.
• 컨슈머 특성: specific.avro.reader=true는 GenericRecord 없이 타입이 지정된 역직렬화를 보장합니다.
• 경쟁 조건 보호: 쓰기는 리더 노드를 통해서만 처리되며, 슬레이브는 요청을 리디렉션합니다.

— Editorial Team