이번 시간에는 Kafka 이벤트에 필드를 추가하거나 지울 때 Schema Registry 호환성 모드가 무엇을 막고 무엇을 놓치는지 확인해 보며 배워가는 시간입니다.
예를 들어서 주문 서비스가 OrderCreated 이벤트에 couponCode 필드를 하나 추가해서 배포했다고 해볼게요.
프로듀서 인스턴스 몇 대는 이미 새 이벤트를 보내고, 나머지는 아직 이전 이벤트를 보내요. 이 토픽을 읽는 정산, 알림, 통계 컨슈머도 각자 다른 시간에 배포되죠.
그러면 토픽에는 이전 형식과 새 형식의 메시지가 섞여 쌓여요. 오래전에 쓴 메시지도 보관 기간 동안 그대로 남아 있고요. 필드 하나를 추가해도, 이 모든 조합에서 컨슈머가 메시지를 읽을 수 있어야 합니다.
Schema Registry는 이런 형식 변경을 등록 단계에서 검사하는 서버예요.
실습 준비
이 글의 명령은 모두 실습 코드의 kafka-schema-registry/ 디렉터리에서 실행합니다.
compose.yaml: 단일 노드 Kafka와 Schema Registry 정의schemas/: 실습에서 등록·검사하는 Avro 스키마(파일 6개)produce.py: 스키마 파일과 JSON 레코드로 메시지 하나를 보내는 프로듀서rawdump.py: 메시지 값을 바이트 그대로 출력하는 컨슈머register.py: Python 클라이언트로 스키마 등록consume.py: 지정한 스키마로 읽고 offset을 commit하는 컨슈머
compose.yaml로 단일 노드 KRaft Kafka와 Schema Registry를 함께 띄울게요. Schema Registry는 스키마를 Kafka 토픽에 저장하기 때문에, 브로커 주소만 알려 주면 됩니다.
SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 스키마를 저장할 Kafka 브로커 주소입니다.SCHEMA_REGISTRY_LISTENERS: REST API를 받을 주소입니다. 호스트에서는 8551 포트로 접속합니다.SCHEMA_REGISTRY_HOST_NAME: 레지스트리 인스턴스를 구분하는 이름입니다.
컨테이너를 띄우고 Python 가상 환경을 만들어 주세요. [avro]를 붙이면 Avro 처리에 쓰는 fastavro와 REST 클라이언트 패키지가 함께 설치돼요.
docker compose -p ksr up -d
python3 -m venv venv && source venv/bin/activate
pip install 'confluent-kafka[avro]==2.15.1'
REST API는 curl로 불러요. 스키마를 파일로 두고 JSON 문자열로 감싸 보내는 일이 반복되니까 셸 함수를 만들어 둘게요.
check는 check <subject> <스키마 파일> [버전 경로]로 부르고, 버전 경로를 생략하면 /latest에 대해 호환성을 검사해요.
아래 블록은 파일로 만들지 말고 실습 디렉터리에서 연 터미널에 그대로 붙여 넣어 주세요. 이후 명령은 모두 이 터미널에서 실행하고, 터미널을 새로 열면 다시 붙여 넣으시면 됩니다.
SR=http://localhost:8551
kafka() { docker compose -p ksr exec -T kafka /opt/kafka/bin/"$@"; }
sr_post() { curl -s -X POST -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
--data "$(jq -n --rawfile s "$2" '{schema: $s}')" "$SR$1"; echo; }
check() { sr_post "/compatibility/subjects/$1/versions${3-/latest}?verbose=true" "$2" \
| jq -r '.is_compatible, (.messages[0] // empty), (.messages[-1] // empty)'; }
register() { sr_post "/subjects/$1/versions" "$2" | jq -c 'del(.schema)'; }
Python 스크립트를 실행할 때 Authlib 패키지가 내는 AuthlibDeprecationWarning 두 줄은 출력에서 뺐어요. 예외가 난 경우에는 트레이스백의 마지막 줄만 남겼습니다.
레지스트리의 기본 호환성 모드부터 확인해 볼게요.
curl -s $SR/config
{"compatibilityLevel":"BACKWARD"}
기본값은 BACKWARD예요.
첫 번째 스키마는 주문 ID, 사용자 ID, 금액만 가진 schemas/order-v1.avsc예요(이하 v1, 파일 이름의 v2~v4도 같은 식으로 부릅니다). 금액은 원 단위 정수입니다.
{"type": "record", "name": "OrderCreated", "namespace": "shop",
"fields": [
{"name": "orderId", "type": "string"},
{"name": "userId", "type": "string"},
{"name": "amount", "type": "long", "doc": "결제 금액(원)"}
]}
orders 토픽을 만들고 REST API로 스키마를 등록해요.
kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic orders --partitions 1
register orders-value schemas/order-v1.avsc
Created topic orders.
{"id":1,"version":1,"guid":"438d05d7-2d20-7f25-9c62-f502fb69481c","schemaType":"AVRO"}
스키마는 subject 단위로 등록해요. subject는 스키마의 버전 이력을 묶는 이름이라고 보시면 됩니다.
클라이언트의 기본 규칙(TopicNameStrategy)은 토픽 이름 뒤에 -value나 -key를 붙여요. 그래서 orders 토픽의 메시지 값 스키마는 orders-value subject에 들어갑니다.
id: 레지스트리 전체에서 스키마 하나를 가리키는 번호입니다. 메시지에 실리는 것은 이 번호입니다.version: subject 안에서 몇 번째 버전인지를 나타냅니다.
프로듀서는 AvroSerializer로 레코드를 바이트로 바꿔 보내요.
auto.register.schemas는 직렬화할 때 스키마를 레지스트리에 자동으로 등록할지 정하는 설정이고, 기본값은 True예요. 여기서는 False로 꺼서 미리 등록한 스키마의 ID만 찾아 쓰게 했어요.
produce.py는 토픽, 스키마 파일, JSON 레코드를 인자로 받아 메시지 하나를 보내고 offset과 값을 출력합니다.
저장된 메시지 값을 가공하지 않고 보려고, 역직렬화하지 않는 컨슈머 rawdump.py도 씁니다. 값의 첫 바이트, 다음 4바이트, 나머지를 나눠 16진수로 출력해요.
주문 하나를 보내고 바이트를 볼게요.
python produce.py orders schemas/order-v1.avsc '{"orderId":"O-1001","userId":"u-1","amount":15000}'
python rawdump.py orders 0 1
offset=0 {'orderId': 'O-1001', 'userId': 'u-1', 'amount': 15000}
offset=0 magic=00 id=00 00 00 01 avro=0c 4f 2d 31 30 30 31 06 75 2d 31 b0 ea 01
- 매직 바이트
00: Schema Registry 직렬화 형식이라는 표시입니다. 지금 형식에서는 항상 0입니다. - 스키마 ID
00 00 00 01: 4바이트 빅엔디언 정수로, 앞에서 등록한 스키마 ID 1입니다. - Avro 본문: 필드 값만 순서대로 들어 있습니다.
여기서 눈여겨볼 건, 본문에 필드 이름도 타입도 없다는 점이에요. 스키마가 없으면 0c가 문자열 길이인지 숫자인지 알 수 없죠.
그래서 컨슈머는 앞의 스키마 ID로 레지스트리에서 스키마를 받아 와야 메시지를 읽을 수 있어요.
실습 1: BACKWARD에서 필드를 추가하면
BACKWARD 모드는 새 스키마로 이전 스키마의 데이터를 읽을 수 있는지 검사해요. 컨슈머를 새 스키마로 올렸을 때, 토픽에 남아 있는 이전 메시지를 읽을 수 있다는 보장이에요.
등록하기 전에 호환성만 따로 검사할 수도 있어요.
POST /compatibility/subjects/orders-value/versions/latest는 후보 스키마를 최신 버전과 비교해 결과만 돌려주고, 아무것도 등록하지 않아요. verbose=true를 붙이면 거절 사유가 함께 옵니다.
후보는 세 가지예요. 모두 v1에 couponCode를 추가하지만 정의가 달라요.
schemas/order-v2.avsc입니다.
{"type": "record", "name": "OrderCreated", "namespace": "shop",
"fields": [
{"name": "orderId", "type": "string"},
{"name": "userId", "type": "string"},
{"name": "amount", "type": "long", "doc": "결제 금액(원)"},
{"name": "couponCode", "type": ["null", "string"], "default": null}
]}
schemas/order-nodefault.avsc입니다.
{"type": "record", "name": "OrderCreated", "namespace": "shop",
"fields": [
{"name": "orderId", "type": "string"},
{"name": "userId", "type": "string"},
{"name": "amount", "type": "long", "doc": "결제 금액(원)"},
{"name": "couponCode", "type": "string"}
]}
schemas/order-nullable.avsc입니다.
{"type": "record", "name": "OrderCreated", "namespace": "shop",
"fields": [
{"name": "orderId", "type": "string"},
{"name": "userId", "type": "string"},
{"name": "amount", "type": "long", "doc": "결제 금액(원)"},
{"name": "couponCode", "type": ["null", "string"]}
]}
check orders-value schemas/order-v2.avsc
check orders-value schemas/order-nodefault.avsc
check orders-value schemas/order-nullable.avsc
true
false
{errorType:'READER_FIELD_MISSING_DEFAULT_VALUE', description:'The field 'couponCode' at path '/fields/3' in the new schema has no default value and is missing in the old schema', additionalInfo:'couponCode'}
{validateFields: 'false', compatibility: 'BACKWARD'}
false
{errorType:'READER_FIELD_MISSING_DEFAULT_VALUE', description:'The field 'couponCode' at path '/fields/3' in the new schema has no default value and is missing in the old schema', additionalInfo:'couponCode'}
{validateFields: 'false', compatibility: 'BACKWARD'}
기본값이 있는 order-v2.avsc만 통과했어요. 왜 그럴까요?
새 스키마로 v1 메시지를 읽는다고 해볼게요. v1 메시지에는 couponCode가 없으니, 그 자리를 채울 기본값이 있어야 해요. 기본값이 없으면 읽을 방법이 없어서 READER_FIELD_MISSING_DEFAULT_VALUE로 거절됩니다.
세 번째 후보는 타입에 null을 넣었는데도 거절됐어요.
Avro에서는 null을 받을 수 있는 타입이라는 것과, 값이 없을 때 null로 채운다는 것이 별개예요. 기본값은 "default": null로 따로 적어 주셔야 합니다.
그리고 Avro 명세에서 union 필드의 기본값은 union의 첫 번째 타입에 맞춰야 해요. 그래서 null 기본값을 쓰려면 ["null", "string"]처럼 null을 앞에 둡니다.
통과한 v2는 이번에는 Python 클라이언트로 등록해 볼게요. register.py는 subject와 스키마 파일을 받아 register_schema로 등록하고, 받은 ID와 최신 버전, 버전 목록을 출력해요.
python register.py orders-value schemas/order-v2.avsc
register_schema -> id 2
latest: subject=orders-value version=2 id=2
versions: [1, 2]
등록할 때도 레지스트리는 같은 호환성 검사를 해요. 검사를 통과하지 못한 스키마는 등록 요청 자체가 거절됩니다.
python produce.py orders schemas/order-v2.avsc '{"orderId":"O-1002","userId":"u-2","amount":32000,"couponCode":"WELCOME"}'
offset=1 {'orderId': 'O-1002', 'userId': 'u-2', 'amount': 32000, 'couponCode': 'WELCOME'}
이제 같은 토픽에 ID 1과 ID 2 메시지가 섞여 있는 거예요.
두 메시지를 컨슈머로 읽어 볼게요. consume.py는 컨슈머가 가진 스키마를 세 번째 인자로 받아요.
AvroDeserializer에 스키마를 주면, 메시지를 쓴 스키마(writer)를 레지스트리에서 받아 온 뒤 컨슈머의 스키마(reader)에 맞춰 변환해요. Java에서 Avro로 생성한 클래스로 읽는 컨슈머가 이렇게 동작합니다. 메시지를 처리한 뒤에는 offset을 commit해요.
v1 스키마를 가진 기존 컨슈머(old-v1 그룹)와 v2 스키마를 가진 새 컨슈머(new-v2 그룹)로 읽어 볼게요.
python consume.py orders old-v1 schemas/order-v1.avsc
python consume.py orders new-v2 schemas/order-v2.avsc
offset=0 {'orderId': 'O-1001', 'userId': 'u-1', 'amount': 15000}
offset=1 {'orderId': 'O-1002', 'userId': 'u-2', 'amount': 32000}
offset=0 {'orderId': 'O-1001', 'userId': 'u-1', 'amount': 15000, 'couponCode': None}
offset=1 {'orderId': 'O-1002', 'userId': 'u-2', 'amount': 32000, 'couponCode': 'WELCOME'}
- 새 컨슈머는 v1 메시지(offset 0)를 읽으면서 없는
couponCode를 기본값None으로 채웠습니다. BACKWARD가 보장하는 방향입니다. - 기존 컨슈머는 v2 메시지(offset 1)를 읽으면서 모르는
couponCode를 버렸습니다. Avro는 reader 스키마에 없는 필드를 무시하므로, 필드 추가는 이 방향으로도 읽힙니다.
다만 BACKWARD가 검사하는 건 첫 번째 방향뿐이에요. 그래서 두 번째 방향이 깨지는 변경도 통과시켜요. 필드를 지울 때 그런 일이 생깁니다.
실습 2: 의미가 바뀐 변경은 통과합니다
예를 들어서 정산팀이 해외 결제를 받으면서, amount를 원 대신 USD 센트로 보내기로 했다고 해볼게요.
타입은 그대로 long이고, 바뀐 건 문서 설명뿐이에요. schemas/order-v3-usd.avsc입니다.
{"type": "record", "name": "OrderCreated", "namespace": "shop",
"fields": [
{"name": "orderId", "type": "string"},
{"name": "userId", "type": "string"},
{"name": "amount", "type": "long", "doc": "결제 금액(USD 센트)"},
{"name": "couponCode", "type": ["null", "string"], "default": null}
]}
check orders-value schemas/order-v3-usd.avsc
register orders-value schemas/order-v3-usd.avsc
python produce.py orders schemas/order-v3-usd.avsc '{"orderId":"O-1003","userId":"u-3","amount":1299,"couponCode":null}'
python consume.py orders old-v1 schemas/order-v1.avsc
true
{"id":3,"version":3,"guid":"fa7989ce-f146-676f-5cfd-3bae7984d35e","schemaType":"AVRO"}
offset=2 {'orderId': 'O-1003', 'userId': 'u-3', 'amount': 1299}
호환성 검사를 통과해서 버전 3으로 등록됐어요. 기존 컨슈머는 오류 없이 amount: 1299를 읽었고, 이 값을 1,299원으로 처리합니다. 실제로는 12.99달러인데 말이죠.
레지스트리가 비교하는 건 필드 이름, 타입, 기본값 같은 구조예요. 값이 무엇을 뜻하는지는 비교하지 않습니다.
단위, 통화, 상태 값의 의미, 타임존 같은 변경은 레지스트리를 그대로 통과해요. 그래서 기존 필드의 의미를 바꾸지 말고 amountUsdCents처럼 이름이 다른 새 필드를 추가해 주세요.
실습 3: 필드를 지우면 기존 컨슈머가 멈춥니다
이번에는 userId를 지운 schemas/order-v4-no-user.avsc를 검사해 볼게요. userId에는 기본값이 없어요.
{"type": "record", "name": "OrderCreated", "namespace": "shop",
"fields": [
{"name": "orderId", "type": "string"},
{"name": "amount", "type": "long", "doc": "결제 금액(USD 센트)"},
{"name": "couponCode", "type": ["null", "string"], "default": null}
]}
check orders-value schemas/order-v4-no-user.avsc
register orders-value schemas/order-v4-no-user.avsc
python produce.py orders schemas/order-v4-no-user.avsc '{"orderId":"O-1004","amount":4500,"couponCode":null}'
true
{"id":4,"version":4,"guid":"69811149-af40-3559-c72d-13bda7760f48","schemaType":"AVRO"}
offset=3 {'orderId': 'O-1004', 'amount': 4500, 'couponCode': None}
BACKWARD는 필드 삭제를 통과시켜요. 새 스키마로 이전 메시지를 읽을 때는 이전 메시지의 userId를 버리면 되기 때문이에요.
그런데 아직 v1 스키마를 가진 기존 컨슈머가 이 메시지를 읽으면 어떻게 될까요?
python consume.py orders old-v1 schemas/order-v1.avsc
fastavro._read_common.SchemaResolutionError: No default value for field userId in shop.OrderCreated
기존 컨슈머는 offset 3에서 예외를 던지고 멈췄어요. v1 스키마에는 userId가 있는데 메시지에는 없고, 채울 기본값도 없기 때문이에요.
정리하면, BACKWARD는 새 컨슈머가 이전 메시지를 읽는 것만 보장하고, 기존 컨슈머가 새 메시지를 읽는 건 보장하지 않아요.
그래서 BACKWARD에서는 컨슈머를 모두 새 스키마로 올린 뒤에 프로듀서를 배포해야 해요. 순서를 거꾸로 하면, 이번 실습처럼 새 메시지가 들어오는 순간 기존 컨슈머가 멈춥니다.
정리
범위: Kafka 4.3.1 단일 노드, Confluent Schema Registry 8.3.2, confluent-kafka[avro] 2.15.1, 파티션 1개인 orders 토픽.
- BACKWARD에서
couponCode추가는"default": null이 있을 때만 통과했고,["null", "string"]타입만 준 후보는READER_FIELD_MISSING_DEFAULT_VALUE로 거절됐어요. 필드를 추가할 때는 기본값을 따로 적고,null기본값이면 union에서null을 앞에 두세요. amount의 단위를 원에서 USD 센트로 바꾼 v3은 검사를 통과했고, 기존 컨슈머는 1299를 원으로 읽었어요. 의미를 바꿀 때는amountUsdCents같은 새 필드나 새 이벤트 타입으로 구조에 드러내 주세요.- BACKWARD는
userId삭제(v4)를 통과시켰고, v1 스키마를 가진 기존 컨슈머는 offset 3에서SchemaResolutionError로 멈췄어요. BACKWARD에서는 컨슈머를 모두 새 스키마로 올린 뒤 프로듀서를 배포하세요. - 이 글에 싣지 않은 추가 측정에서는, 반대 방향(이전 스키마로 새 데이터를 읽을 수 있는지)을 검사하는 FORWARD가 기본값 없는 필드 삭제를 등록 단계에서 거절했고, 두 방향을 모두 검사하는 FULL에서는 기본값이 있는/없는 필드의 추가와 삭제 네 후보 중 기본값이 있는 필드의 추가와 삭제만 통과했어요. FORWARD에서는 프로듀서를 먼저 배포하고, 배포 순서를 가리지 않으려면 추가하거나 지울 필드에 기본값을 두세요.
- 같은 추가 측정에서
userId를 long으로 다시 넣은 스키마는 최신 버전(v4)과만 비교하는 BACKWARD 검사를 통과했지만, 토픽을 처음부터 읽자string is not long으로 실패했어요. 등록된 모든 버전과 비교하는BACKWARD_TRANSITIVE로 바꿔도/versions/latest로 검사하면true였어요. 오래된 메시지를 다시 읽는다면_TRANSITIVE모드를 쓰고/versions로 검사하세요.