🏳️‍🌈 [궁금한점]

• schema-registry 설치 방법 및 테스트 방법

---

🔗목차

• kafka 브로커 실행
  • 도커를 이용한 kafka 실행 (Standalone)
  • Docker Run 명령어 옵션 설명
• kafka 브로커 테스트
  • kafka 컨테이너 접속
  • 자바 버전 확인
  • Kafka 버전 확인
  • 토픽 생성
  • 토픽 목록 조회
  • 토픽 삭제
  • 토픽 설정 조회
  • 프로듀서
  • 컨수머
  • 소비자 그룹 조회
• 클라이언트 구성
• 백업
  • MirrorMaker 2 (MM2)
  • Kafka Connect + Sink Connector
  • Kafka Consumer 스크립트로 직접 Export
  • Storage Level Backup (디스크 기반)
• schema-registry 설치
  • 네트워크 생성
  • Zookeeper 설치
  • schema-registry 설치
  • schema-registry 확인
• schema-registry 테스트
  • topic에 스키마 등록 (Avro 예제)
  • 스키마 목록 조회
  • 특정 주제의 최신 스키마 조회
  • 특정 버전의 스키마 조회
  • 등록된 스키마 삭제
  • 스키마 호환성 확인
• schema-registry 사용방법
  • 메시지 전송
  • 스키마 변경 발생
  • 스키마 등록

kafka 브로커 실행

도커를 이용한 kafka 실행 (Standalone)

docker run -d  \
  --name kafka \
  -p 9092:9092 \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
  -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
  -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
  -e KAFKA_NUM_PARTITIONS=3 \
  apache/kafka:latest

Docker Run 명령어 옵션 설명

옵션	설명
-d	-d: 컨테이너를 백그라운드에서 실행
--name kafka	컨테이너 이름을 kafka로 설정
-e KAFKA_NODE_ID=1	Kafka 노드의 ID를 1로 설정. Kafka 클러스터 내에서 고유한 ID로 사용됩니다.
-e KAFKA_PROCESS_ROLES=broker,controller	Kafka가 broker와 controller 역할을 동시에 수행하도록 설정.
-e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093	Kafka가 PLAINTEXT와 CONTROLLER 리스너로 각각 9092와 9093 포트에서 리스닝하도록 설정
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092	외부에서 접근할 수 있는 Kafka 브로커의 주소를 localhost:9092로 설정
-e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER	Kafka 컨트롤러가 사용할 리스너의 이름을 CONTROLLER로 설정.
-e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT	리스너에 대해 사용할 보안 프로토콜을 설정.
-e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093	Kafka 컨트롤러가 사용할 쿼럼 노드를 설정.
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1	오프셋 토픽의 복제 계수를 1로 설정하여 복제본을 1개로 제한합니다.
-e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1	트랜잭션 상태 로그의 복제 계수를 1로 설정하여 복제본을 1개로 제한합니다.
-e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1	트랜잭션 상태 로그의 최소 ISR(In-Sync Replicas)을 설정.
-e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0	소비자 그룹의 초기 리밸런스를 즉시 수행하도록 지연 시간을 0으로 설정.
-e KAFKA_NUM_PARTITIONS=3	기본적으로 생성되는 Kafka 토픽의 파티션 수를 3으로 설정.
apache/kafka:latest	Docker에서 실행할 Kafka의 최신 버전 이미지를 사용합니다.

kafka 브로커 테스트

kafka 컨테이너 접속

docker exec -it kafka bash

cf9fccdc5f12:/var/lib/kafka/data$ export PATH=$PATH:/opt/kafka/bin/

자바 버전 확인

8938869c94ce:/$ java --version
openjdk 21.0.6 2025-01-21 LTS
OpenJDK Runtime Environment Temurin-21.0.6+7 (build 21.0.6+7-LTS)
OpenJDK 64-Bit Server VM Temurin-21.0.6+7 (build 21.0.6+7-LTS, mixed mode, sharing)

Kafka 버전 확인

26d922b2998b:/$ kafka-topics.sh --version
4.0.0

토픽 생성

cf9fccdc5f12:~$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test-topic
Created topic test-topic.

cf9fccdc5f12:~$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test-topic2
Created topic test-topic2.

토픽 목록 조회

cf9fccdc5f12:~$ kafka-topics.sh --list --bootstrap-server localhost:9092
__consumer_offsets
_schemas
test-topic
test-topic2

토픽 삭제

cf9fccdc5f12:~$ kafka-topics.sh --delete --topic test-topic2 --bootstrap-server localhost:9092

cf9fccdc5f12:~$ kafka-topics.sh --list --bootstrap-server localhost:9092
__consumer_offsets
_schemas
test-topic

토픽 설정 조회

cf9fccdc5f12:~$ kafka-configs.sh --describe --topic test-topic --bootstrap-server localhost:9092
Dynamic configs for topic test-topic are:

프로듀서

cf9fccdc5f12:~$ kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic
>

컨수머

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning

소비자 그룹 조회

kafka-consumer-groups.sh --list --bootstrap-server localhost:9092
console-consumer-73216
console-consumer-34026

클라이언트 구성

acho@DESKTOP-SCOK45O:~$ rm kafka_2.12-3.7.2.tgz
acho@DESKTOP-SCOK45O:~$ tar -xvf kafka_2.13-4.0.0.tgz

백업

Kafka의 백업은 전통적인 데이터베이스처럼 "스냅샷"을 찍는 방식보다는 topic 데이터를 복제하거나 내보내는(export) 방식으로 수행하는 게 일반적이다. 아래는 Kafka 백업을 수행할 수 있는 대표적인 방법들이다

MirrorMaker 2 (MM2)

Kafka의 데이터를 다른 Kafka 클러스터로 복제(미러링)할 수 있는 공식 도구이다.
실시간 복제, 장애 복구 클러스터 구성 가능하며, DR(재해 복구)나 다른 지역으로 동기화할 때 사용한다.

Kafka Connect + Sink Connector

Kafka Connect 프레임워크를 사용해 데이터를 외부 저장소로 내보낼 수 있다.

• S3 Sink Connector: AWS S3나 MinIO에 백업
• HDFS Sink Connector
• Elasticsearch, PostgreSQL, BigQuery 등

다양한 외부 시스템 지원, 플러그인 형식이 존재하나 Kafka Connect 클러스터 사용이 요구된다

Kafka Consumer 스크립트로 직접 Export

간단한 Python이나 Java 스크립트를 사용하여 데이터를 직접 읽어서 파일로 저장할 수 있다.

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'your-topic',
    bootstrap_servers='kafka:9092',
    auto_offset_reset='earliest',
    enable_auto_commit=False
)

with open('backup.json', 'w') as f:
    for message in consumer:
        f.write(f"{message.value.decode('utf-8')}\n")

커스터마이징 쉬운 반면 대량 데이터에는 비효율적이다.

Storage Level Backup (디스크 기반)

Kafka의 로그 디렉토리(/var/lib/kafka 등)를 직접 백업하는 방식이다. kafka.log.dirs에 저장되는 .log 파일을 tar, rsync, snapshot 등으로 복사한다.

이 방법은 Kafka 브로커를 중지한 후 백업해야 일관성 보장된다.

schema-registry 설치

네트워크 생성

docker network create kafka-net

Zookeeper 설치

docker run -d \
--network kafka-net \
  --name zookeeper \
  -e ZOOKEEPER_CLIENT_PORT=2181 \
  confluentinc/cp-zookeeper:7.5.0

schema-registry 설치

docker run -d \
  --name schema-registry \
  --network kafka-net \
  -p 8081:8081 \
  -e SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS=PLAINTEXT://kafka:9092 \
  -e SCHEMA_REGISTRY_HOST_NAME=schema-registry \
  -e SCHEMA_REGISTRY_LISTENERS=http://0.0.0.0:8081 \
  confluentinc/cp-schema-registry:7.5.0

schema-registry 확인

acho@DESKTOP-SCOK45O:~curl -X POST http://localhost:8081/subjects/my-topic-value/versions \ \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"}]}"}'

{"id":1}

schema-registry 테스트

topic에 스키마 등록 (Avro 예제)

acho@DESKTOP-SCOK45O:~$ curl -X POST http://localhost:8081/subjects/test-topic-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{
    "schema": "{\"type\": \"record\", \"name\": \"User\", \"fields\": [{\"name\": \"id\", \"type\": \"int\"}, {\"name\": \"name\", \"type\": \"string\"}]}"
  }'

{"id":2}

• test-topic-value: value에 대한 스키마. key라면 test-topic-key
• schema: Avro JSON 형식 (이중 인코딩에 주의)

스키마 목록 조회

acho@DESKTOP-SCOcurl http://localhost:8081/subjectssubjects

["my-topic-value","test-topic-value"]

특정 주제의 최신 스키마 조회

curl http://localhost:8081/subjects/test-topic-value/versions/latest

특정 버전의 스키마 조회

acho@DESKTOP-SCOK45O:~$ curl http://localhost:8081/subjects/test-topic-value/versions/1

{"subject":"test-topic-value","version":1,"id":2,"schema":"{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"},{\"name\":\"name\",\"type\":\"string\"}]}"}

등록된 스키마 삭제

curl -X DELETE http://localhost:8081/subjects/test-topic-value

스키마 호환성 확인

curl -X POST http://localhost:8081/compatibility/subjects/test-topic-value/versions/latest \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{
    "schema": "{\"type\": \"record\", \"name\": \"User\", \"fields\": [{\"name\": \"id\", \"type\": \"int\"}]}"
  }'

schema-registry 사용방법

메시지 전송

echo '{"name": "Alice"}' | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic

• 정상 전송

스키마 변경 발생

echo '{"name": "Alice", "age": 30}' | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic

• 컨슈머가 구조 변경 감지 못 해 → 오류 발생 가능성 있음.

스키마 등록

acho@DESKTOP-SCOK45O:~$ curl -X POST http://localhost:8081/subjects/test-topic-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{
    "schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"}]}"
  }'
{"id":1}