이번 시간에는 같은 파티션 안에서도 메시지 처리 순서가 깨지는 두 경우를 직접 재현해 보고, 순서를 지키려면 무엇을 정해야 하는지 배워가는 시간입니다.
예를 들어서 주문 서비스가 order-1001 주문에 대해 created, paid, shipped 이벤트를 차례로 보냈다고 해볼게요. 배송 서비스는 주문 번호를 키로 받기 때문에, 세 이벤트는 같은 파티션에 순서대로 쌓였어요. 그런데 배송 서비스의 주문 테이블에는 이 주문이 paid 상태로 남았습니다. 어떻게 된 걸까요?
Kafka가 지키는 것은 메시지가 파티션에 저장된 순서이고, 컨슈머는 그 순서대로 메시지를 받아요. 하지만 받은 메시지를 서비스가 어떤 순서로 처리해서 끝내는지는 Kafka가 관여하지 않습니다. 게다가 저장 순서도 프로듀서 설정에 따라 보낸 순서와 달라질 수 있어요.
실습 준비
이 글의 명령은 모두 실습 코드의 kafka-message-ordering/ 디렉터리에서 실행합니다.
compose.yaml: 리스너 3개(PLAINTEXT,HOST,FAULT)를 둔 KRaft 단일 노드 Kafkaorders.txt: 주문 3건의 이벤트 9건을 섞어 적은 입력 파일worker.py: 메시지 9건을 세 가지 처리 방식 중 하나로 처리하고 주문별 최종 상태를 출력하는 컨슈머proxy.py: 두 번째 Produce 요청 하나만 브로커에 넘기지 않고 버리는 테스트용 프록시producer.py:order-1001의 이벤트 3건을 프로듀서 설정을 바꿔 가며 보내는 프로듀서
리스너는 용도에 따라 나눠 두었어요.
PLAINTEXT(kafka:9092): 컨테이너 안에서 Kafka 명령 도구가 써요.HOST(localhost:9461): 호스트의 Python 컨슈머가 씁니다.FAULT(localhost:9462): 프로듀서 재시도 실습에서 써요. 호스트의 9462 포트에는 테스트용 프록시가 뜨고, 프록시가 컨테이너의 9463 포트로 요청을 넘깁니다.
compose.yaml은 이렇게 생겼어요.
services:
kafka:
image: apache/kafka:4.3.1
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,HOST://:9461,FAULT://:9463
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,HOST://localhost:9461,FAULT://localhost:9462
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,HOST:PLAINTEXT,FAULT:PLAINTEXT
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_HEAP_OPTS: -Xmx512m -Xms256m
ports:
- "9461:9461"
- "9463:9463"
컨테이너를 띄우고, Kafka 명령을 짧게 쓰기 위한 셸 함수를 만들어 놓을게요.
아래 블록은 파일로 만들지 말고 실습 디렉터리에서 연 터미널에 그대로 붙여 넣어 주세요. 이후 명령은 모두 이 터미널에서 실행하고, 터미널을 새로 열면 다시 붙여 넣으시면 됩니다.
docker compose up -d
kafka() { docker compose exec -T kafka /opt/kafka/bin/"$@"; }
python3 -m venv venv && source venv/bin/activate
pip install confluent-kafka
파티션이 1개인 orders 토픽을 만들고 orders.txt의 이벤트를 넣을게요. orders.txt에는 주문 3건의 이벤트를 섞어서 적었고, : 앞이 키(주문 번호), 뒤가 이벤트입니다.
kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic orders --partitions 1
kafka kafka-console-producer.sh --bootstrap-server kafka:9092 --topic orders \
--reader-property parse.key=true --reader-property key.separator=: < orders.txt
kafka kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic orders --from-beginning --max-messages 9 \
--formatter-property print.offset=true --formatter-property print.key=true 2> /dev/null
Created topic orders.
Offset:0 order-1001 created
Offset:1 order-1002 created
Offset:2 order-1001 paid
Offset:3 order-1003 created
Offset:4 order-1002 paid
Offset:5 order-1001 shipped
Offset:6 order-1003 paid
Offset:7 order-1002 shipped
Offset:8 order-1003 shipped
보낸 순서 그대로 offset(파티션 안에서 0부터 늘어나는 위치 번호) 0부터 8까지 저장됐고, 컨슈머도 이 순서대로 받아요. Kafka가 책임지는 범위는 여기까지입니다.
실습 1: 스레드 풀에 넘기면 늦게 받은 메시지가 먼저 끝납니다
배송 서비스 컨슈머는 worker.py예요. 메시지 9개를 받아서 처리하고, 처리가 끝난 시간과 주문별 최종 상태를 출력합니다. 처리 방식은 인자로 고를 수 있어요.
serial: 한 스레드가 받은 순서대로 하나씩 처리합니다.pool: 스레드 4개짜리 스레드 풀에 메시지를 넘겨 동시에 처리해요.key: 키마다 정해진 실행기(스레드 하나)에서 처리합니다. 실습 2에서 써요.
paid는 결제 API를 호출한다고 보고 0.5초, 나머지는 0.1초가 걸립니다. 처리 결과는 status 딕셔너리에 주문별 마지막 상태로 덮어써요. 실제 서비스라면 주문 테이블의 상태 컬럼을 갱신하는 자리라고 보시면 됩니다.
한 스레드로 처리한 결과와 스레드 풀로 처리한 결과를 차례로 볼게요.
python worker.py serial
python worker.py pool
0.1s 완료 offset=0 order-1001 created
0.2s 완료 offset=1 order-1002 created
0.7s 완료 offset=2 order-1001 paid
0.8s 완료 offset=3 order-1003 created
1.3s 완료 offset=4 order-1002 paid
1.4s 완료 offset=5 order-1001 shipped
1.9s 완료 offset=6 order-1003 paid
2.0s 완료 offset=7 order-1002 shipped
2.1s 완료 offset=8 order-1003 shipped
걸린 시간 2.2s, 최종 상태 {'order-1001': 'shipped', 'order-1002': 'shipped', 'order-1003': 'shipped'}
0.1s 완료 offset=1 order-1002 created
0.1s 완료 offset=0 order-1001 created
0.1s 완료 offset=3 order-1003 created
0.2s 완료 offset=5 order-1001 shipped
0.3s 완료 offset=7 order-1002 shipped
0.4s 완료 offset=8 order-1003 shipped
0.5s 완료 offset=2 order-1001 paid
0.6s 완료 offset=4 order-1002 paid
0.6s 완료 offset=6 order-1003 paid
걸린 시간 0.6s, 최종 상태 {'order-1001': 'paid', 'order-1002': 'paid', 'order-1003': 'paid'}
한 스레드로 처리하면 offset 순서대로 끝나 세 주문 모두 shipped예요. 대신 처리 시간을 모두 더한 2.2초가 걸렸습니다.
스레드 풀은 0.6초로 줄었지만 결과가 틀렸어요. order-1001을 보면 offset 5의 shipped가 0.2초에 끝났고, 그보다 앞선 offset 2의 paid는 0.5초에 끝났습니다. 늦게 끝난 paid가 상태를 덮어써서 세 주문 모두 최종 상태가 paid로 남았어요.
처리 시간이 다르면 끝나는 순서는 넘긴 순서와 상관없어지기 때문이에요. Kafka가 지켜 주는 것은 전달 순서까지이고, 처리 완료 순서는 컨슈머 코드가 지켜야 합니다. asyncio 태스크처럼 결과를 기다리지 않고 다음 메시지로 넘어가거나, 내부 큐에 넣고 여러 워커가 꺼내 가는 경우에도 같은 일이 생길 수 있어요.
실습 2: 키마다 실행기를 정하면 순서와 속도를 같이 얻습니다
배송 서비스에서 순서가 필요한 단위는 주문 하나예요. order-1001의 paid와 shipped는 차례대로 끝나야 하지만, order-1001과 order-1002는 서로 기다릴 이유가 없죠.
key 방식은 스레드 1개짜리 실행기 4개를 써요. 키의 crc32 해시를 실행기 수로 나눈 나머지로 실행기를 고르니까, 같은 키는 늘 같은 실행기로 갑니다. 출력의 키별 줄은 키마다 고른 실행기 번호예요.
commit(다음에 읽을 offset을 컨슈머 그룹 이름으로 저장하는 일)은 받은 메시지를 모두 처리한 뒤에 한 번 합니다.
python worker.py key
kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --group shipping-key 2> /dev/null
키별 줄: {'order-1001': 1, 'order-1002': 3, 'order-1003': 1}
0.1s 완료 offset=0 order-1001 created
0.1s 완료 offset=1 order-1002 created
0.6s 완료 offset=2 order-1001 paid
0.6s 완료 offset=4 order-1002 paid
0.7s 완료 offset=7 order-1002 shipped
0.7s 완료 offset=3 order-1003 created
0.8s 완료 offset=5 order-1001 shipped
1.3s 완료 offset=6 order-1003 paid
1.4s 완료 offset=8 order-1003 shipped
걸린 시간 1.4s, 최종 상태 {'order-1001': 'shipped', 'order-1002': 'shipped', 'order-1003': 'shipped'}
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
shipping-key orders 0 9 9 0 - - -
세 주문 모두 최종 상태가 shipped이고, 주문마다 created, paid, shipped 순서로 끝났어요. 전체 시간은 1.4초로, 한 스레드로 처리한 2.2초보다 짧습니다. 그룹의 offset도 9까지 올라갔어요.
다만 order-1001과 order-1003이 같은 실행기 1에 들어갔죠. 그래서 order-1003의 created는 order-1001의 세 이벤트가 끝날 때까지 기다렸다가 0.7초에야 끝났어요. 키 종류에 비해 실행기가 너무 적거나 이벤트가 몰리는 키가 있으면, 그 실행기에만 처리할 이벤트가 쌓여요.
실습 3: 프로듀서 재시도로 저장 순서가 바뀌는 경우
프로듀서는 응답을 기다리지 않고 요청을 여러 개 보낼 수 있어요. 응답을 아직 못 받은 요청을 in-flight 요청이라고 하고, 연결 하나에 몇 개까지 허용할지를 max.in.flight.requests.per.connection으로 정합니다.
Kafka 문서는 이 값이 1보다 크고 멱등성 설정 enable.idempotence(재시도로 같은 메시지가 두 번 저장되지 않게 하는 기능)가 꺼져 있으면, 전송 실패 뒤 재시도 때문에 메시지 순서가 바뀔 위험이 있다고 설명해요. librdkafka 기반 클라이언트(Python confluent-kafka 등)는 enable.idempotence 기본값이 false, max.in.flight.requests.per.connection 기본값이 1000000이라 기본값 그대로 쓰면 이 조건에 해당합니다.
재시도가 일어나려면 요청이 실패해야 해서, 프로듀서와 브로커 사이에 작은 프록시를 두는 방식으로 가볼게요. proxy.py는 두 번째 Produce 요청 하나만 브로커에 넘기지 않고 버려요. 호스트의 9462 포트에서 받아 컨테이너의 9463 포트로 넘깁니다.
producer.py는 order-1001의 세 이벤트를 보내요. created를 보내고 응답까지 받은 뒤, paid를 보내고 1초 기다렸다가 shipped를 보냅니다. 프록시가 버리는 요청이 paid예요. socket.timeout.ms는 3초로 줄여 두어서 librdkafka가 응답 없는 요청을 3초 만에 실패로 보고 다시 보냅니다.
아래 produce_case <토픽> <멱등성> <in-flight 수>가 토픽 만들기, 보내기, 저장 순서 확인을 한 번에 해 줍니다. 멱등성을 끈 경우와 켠 경우를 in-flight 5로 차례로 보내 볼게요.
produce_case() {
kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic "$1" --partitions 1
python proxy.py & local px=$!
sleep 1
python producer.py "$@"
kill $px
kafka kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic "$1" --from-beginning --max-messages 3 \
--formatter-property print.offset=true 2> /dev/null
}
produce_case retry-a false 5
produce_case retry-c true 5
Created topic retry-a.
0.0s created 전송 결과: offset=0
proxy: Produce 요청 correlation_id=3 버림
1.0s shipped 전송 결과: offset=1
%5|1790813018.166|REQTMOUT|rdkafka#producer-1| [thrd:localhost:9462/1]: localhost:9462/1: Timed out ProduceRequest in flight (after 3038ms, timeout #0)
%4|1790813018.166|REQTMOUT|rdkafka#producer-1| [thrd:localhost:9462/1]: localhost:9462/1: Timed out 1 in-flight, 0 retry-queued, 0 out-queue, 0 partially-sent requests
3.2s paid 전송 결과: offset=2
Offset:0 created
Offset:1 shipped
Offset:2 paid
Created topic retry-c.
0.5s created 전송 결과: offset=0
proxy: Produce 요청 correlation_id=4 버림
%5|1790813047.092|REQTMOUT|rdkafka#producer-1| [thrd:localhost:9462/1]: localhost:9462/1: Timed out ProduceRequest in flight (after 3041ms, timeout #0)
%4|1790813047.092|REQTMOUT|rdkafka#producer-1| [thrd:localhost:9462/1]: localhost:9462/1: Timed out 1 in-flight, 0 retry-queued, 0 out-queue, 0 partially-sent requests
3.7s paid 전송 결과: offset=1
3.7s shipped 전송 결과: offset=2
Offset:0 created
Offset:1 paid
Offset:2 shipped
멱등성을 끈 retry-a에서는 in-flight가 5까지 허용되니까 프로듀서가 paid의 응답을 기다리지 않고 1초 뒤 shipped를 보냈고, shipped가 offset 1에 저장됐습니다. 3초가 지나자 librdkafka가 paid 요청을 타임아웃으로 처리하고(REQTMOUT 로그) 다시 보내서 paid는 offset 2에 저장됐어요.
파티션 하나, 키 하나인데 shipped가 paid보다 앞에 저장된 거예요. 이러면 컨슈머가 한 스레드로 offset 순서를 지켜 처리해도 결과는 틀립니다.
멱등성을 켠 retry-c는 in-flight가 5 그대로인데 순서가 지켜졌어요. 멱등성을 켜면 프로듀서는 브로커에서 프로듀서 ID를 받고, 파티션마다 배치에 0부터 늘어나는 순번(sequence number)을 붙여요. 브로커는 프로듀서마다 다음에 받을 순번을 기억하고, 순번이 건너뛴 배치는 저장하지 않고 거절합니다. 그래서 shipped가 먼저 저장되지 못하고, 프로듀서가 paid와 shipped를 원래 순서대로 다시 보냈어요.
멱등 프로듀서가 재시도 중복을 어떻게 막는지 더 궁금하면 멱등 프로듀서를 다룬 글을 참고하세요.
정리
범위: Kafka 4.3.1(KRaft 단일 노드), Python confluent-kafka 2.15.1(librdkafka 2.15), 파티션 1개 토픽, 주문 3건의 이벤트 9건, paid 처리 0.5초와 나머지 0.1초.
- 한 스레드로 처리하자 offset 순서대로 끝나 세 주문 모두
shipped였지만, 처리 시간을 모두 더한 2.2초가 걸렸습니다. - 스레드 4개 풀은 0.6초로 줄었지만 offset 5의
shipped가 offset 2의paid보다 먼저 끝나 세 주문 모두paid로 남았습니다. 상태를 덮어쓰는 처리라면 같은 키를 제한 없이 병렬로 처리하지 마세요. - 키의 crc32로 실행기 4개 중 하나를 고르자 주문마다 순서가 지켜지고 1.4초가 걸렸지만, 같은 실행기에 들어간
order-1003의created는order-1001을 기다려 0.7초에 끝났습니다. 실행기 수는 키 종류와 이벤트가 몰리는 키를 보고 정하세요. - 멱등성을 끄고 in-flight를 5로 두자 응답 없는
paid가 3초 뒤 재시도되는 사이shipped가 offset 1에 먼저 저장됐습니다. 순서가 필요한 토픽에 보내는 프로듀서는enable.idempotence를 켜 주세요. - 멱등성을 켜자 in-flight 5에서도
paid,shipped순서로 저장됐습니다.