Kafka 리밸런스가 끝나면 왜 같은 메시지를 또 처리할까?

이번 시간에는 Kafka 컨슈머 리밸런스 때 같은 메시지를 또 처리하는 상황을 직접 재현해 보며 배워가는 시간입니다.

예를 들어서 우리가 배포하려고 컨슈머를 하나씩 재시작했을 뿐인데, 같은 주문이 수십 건씩 두 번 처리됐다고 해볼게요.

리밸런스는 그룹 구성원이 바뀔 때 파티션 담당자를 다시 정하는 과정이에요. classic 프로토콜은 리밸런스 때 구성원이 모두 그룹에 다시 참여하고, 구성원 중 리더가 할당을 정하는 기존 방식이에요.

commit은 다음에 읽을 offset을 저장해요. 그래서 commit 전에 컨슈머가 멈추면, 마지막 commit 뒤에 처리한 메시지는 다음 담당자가 다시 처리합니다. 이 기본 동작이 더 궁금하면 offset을 다룬 글을 참고하세요.

재현해 볼 건 다음 두 가지예요.

• 파티션을 반납할 때 호출되는 콜백 on_revoke에서 commit하지 않으면 중복이 몇 건 생기는지
• 처리가 max.poll.interval.ms를 넘기면 컨슈머와 commit이 어떻게 되는지

실습 준비

이 글의 명령은 모두 실습 코드의 kafka-rebalance/ 디렉터리에서 실행합니다.

• compose.yaml: 단일 브로커 Kafka 4.3
• common.sh: 토픽 생성, 컨슈머 시작·정지 셸 함수
• worker.py: 모든 실습에 쓰는 수동 commit 컨슈머
• produce.py: 파티션을 돌아가며 메시지를 넣는 프로듀서
• dups.py: 로그에서 같은 파티션·offset의 중복 처리를 세는 스크립트
• 01-revoke.sh: 실습 1 실행 스크립트
• 02-maxpoll.sh: 실습 2 실행 스크립트

Docker로 단일 브로커 Kafka 4.3을 띄울게요. 호스트의 Python 컨슈머는 localhost:9481로 접속해요. 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://:9481
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,HOST://localhost:9481
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,HOST: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_HEAP_OPTS: -Xmx512m -Xms256m
    ports:
      - "9481:9481"

Python 가상 환경은 작업 디렉터리 바로 위 디렉터리에 만들어 주세요.

docker compose -p kreb up -d
python3 -m venv ../venv && ../venv/bin/pip install confluent-kafka==2.15.1

반복해서 쓰는 셸 함수는 common.sh에 모아 두었어요. 이 파일은 logs/ 디렉터리를 만들고, Python은 ../venv/bin/python을 씁니다. 셸에서 직접 명령을 실행할 때도 source common.sh를 먼저 실행해 주세요.

• topic: 지정한 파티션 수로 토픽을 만듭니다.
• start: 컨슈머를 백그라운드로 띄우고 로그를 logs/<이름>.log에 남깁니다.
• stop: SIGTERM을 보내 정상 종료시킵니다.

모든 실습에 쓰는 컨슈머는 worker.py이고, start <이름> <그룹> <토픽> [옵션]으로 띄워요. 자동 commit을 끄고, 처리한 메시지 수가 --commit-every에 이를 때마다 직접 commit합니다.

• --delay: 메시지 하나를 처리하는 데 걸리는 시간(초)입니다.
• --commit-every: 메시지 몇 개마다 commit할지 정합니다. 기본 100입니다.
• --revoke-commit: 파티션을 반납할 때 on_revoke에서 처리한 위치까지 commit합니다.
• --slow-at, --slow-sec: 지정한 offset에서 처리 시간을 길게 만듭니다.

on_lost는 파티션을 정상적으로 반납하지 못하고 잃었을 때 호출되는 콜백이에요.

메시지는 produce.py <토픽> <개수> <파티션 수> [간격 초]로 보내요. 중복 처리 건수는 dups.py로 셉니다. 로그에서 같은 파티션, 같은 offset이 두 번 이상 나온 것을 세요.

실습 1: on_revoke에서 commit하지 않으면 중복이 생깁니다

파티션 2개짜리 pay 토픽에 메시지 1000개를 넣고, 컨슈머 a 혼자 처리하게 해 볼게요. 메시지 하나에 0.01초가 걸리고 100개마다 commit합니다.

a가 처리하는 도중에 b를 추가해서 리밸런스를 일으켜요. 01-revoke.sh <그룹> [--revoke-commit]이 이 과정을 실행하고, 1000개를 모두 처리하면 두 컨슈머를 멈춥니다.

먼저 on_revoke에서 아무것도 하지 않는 nocommit 그룹이에요.

source common.sh
chmod +x 01-revoke.sh
topic pay 2
$PY produce.py pay 1000 2
./01-revoke.sh nocommit
$PY dups.py nocommit

Created topic pay.
sent 1000 messages to pay
09:13:27.038 start nocommit-b
processed 1082, unique 1000, duplicates 82
  p1: offset 400..481 (82) processed twice

메시지 1000개를 처리했어야 하는데, 처리 기록은 1082건이에요. 파티션 1의 offset 400부터 481까지 82건이 두 번 처리됐습니다.

왜 그럴까요? 리밸런스 직전 부분의 로그를 볼게요.

cat logs/nocommit-*.log | sort | awk '$1>="09:13:29.82" && $1<="09:13:29.90"'

09:13:29.829 nocommit-a p1 offset 479
09:13:29.841 nocommit-a p1 offset 480
09:13:29.854 nocommit-a p1 offset 481
09:13:29.854 nocommit-a revoked [0, 1]
09:13:29.861 nocommit-a assigned [0]
09:13:29.861 nocommit-b assigned [1]
09:13:29.876 nocommit-b p1 offset 400
09:13:29.889 nocommit-b p1 offset 401

a는 파티션 1을 481까지 처리하고 반납했지만, 그룹에 저장된 위치는 400이었어요. 그래서 파티션 1을 넘겨받은 b는 400부터 읽었고, 장애가 없었는데도 82건이 중복됐습니다.

중복 건수는 마지막 commit 이후 처리한 양과 같아요. 100개마다 commit하니까 리밸런스 시점에 따라 0건에서 99건 사이로 달라집니다.

이번에는 --revoke-commit을 붙인 revcommit 그룹으로 같은 과정을 반복해 볼게요.

./01-revoke.sh revcommit --revoke-commit
$PY dups.py revcommit
cat logs/revcommit-*.log | sort | awk '$1>="09:13:56.04" && $1<="09:13:56.10"'

09:13:53.132 start revcommit-b
processed 1000, unique 1000, duplicates 0
09:13:56.042 revcommit-a p1 offset 487
09:13:56.053 revcommit-a p1 offset 488
09:13:56.066 revcommit-a p1 offset 489
09:13:56.066 revcommit-a revoked [0, 1]
09:13:56.072 revcommit-a commit p1=490
09:13:56.079 revcommit-a assigned [0]
09:13:56.079 revcommit-b assigned [1]
09:13:56.098 revcommit-b p1 offset 490

a가 반납 직전에 p1=490을 commit했고, b는 490부터 이어서 읽었어요. 중복은 0건입니다.

on_revoke에서 commit할 때는 지켜 주셔야 할 점이 있어요.

• 동기 commit을 씁니다. 비동기로 보내 놓고 콜백을 빠져나가면 commit이 끝나기 전에 새 담당자가 읽기 시작할 수 있습니다.
• 콜백 안에서 오래 머물지 않습니다. classic 프로토콜은 모든 구성원이 반납을 끝내야 다음 단계로 넘어가므로, 콜백이 느리면 그룹 전체가 그만큼 기다립니다.

다만 이 방법은 정상적인 리밸런스에서 생기는 중복만 줄여 줘요.

실습 2: 처리가 느린 컨슈머는 그룹에서 쫓겨납니다

컨슈머가 poll()을 너무 오래 부르지 않아도 리밸런스가 일어나요. 이때는 클라이언트가 스스로 그룹을 떠납니다.

컨슈머에는 살아 있는지 확인하는 타이머가 두 개 있어요.

• session.timeout.ms: heartbeat가 이 시간 동안 오지 않으면 코디네이터가 컨슈머를 그룹에서 뺍니다. heartbeat는 클라이언트의 백그라운드 스레드가 보내므로, 처리가 오래 걸려도 프로세스만 살아 있으면 계속 갑니다. confluent-kafka와 Java 클라이언트 모두 기본값은 45000(45초)입니다.
• max.poll.interval.ms: poll()을 부른 뒤 다음 poll()까지 허용하는 최대 시간입니다. 넘기면 클라이언트가 처리가 멈췄다고 보고 그룹을 떠납니다. 기본값은 300000(5분)입니다.

파티션 1개짜리 slow 토픽에 메시지 10개를 넣습니다. a와 b는 같은 그룹이지만 파티션이 하나라서 a만 일하고 b는 대기해요.

메시지 하나에 0.5초가 걸리고 매번 commit합니다. a는 offset 3에서 15초가 걸리게 하고, max.poll.interval.ms는 10초로 줄였어요.

confluent-kafka는 max.poll.interval.ms가 session.timeout.ms보다 작으면 컨슈머를 만들지 않아요. 그래서 session.timeout.ms도 브로커가 허용하는 최솟값인 6초로 줄였습니다. 02-maxpoll.sh가 이 과정을 실행해요.

chmod +x 02-maxpoll.sh
topic slow 1
./02-maxpoll.sh
cat logs/slow-*.log | sort | awk '$1 < "09:15"'

%4|1790813686.528|MAXPOLL|slow-a#consumer-1| [thrd:main]: Application maximum poll interval (10000ms) exceeded by 356ms (adjust max.poll.interval.ms for long-running message processing): leaving group
09:14:31.388 slow-a assigned [0]
09:14:34.419 slow-a revoked [0]
09:14:34.425 slow-a assigned [0]
09:14:34.425 slow-b assigned []
09:14:35.157 slow-a p0 offset 0
09:14:35.161 slow-a commit p0=1
09:14:35.663 slow-a p0 offset 1
09:14:35.666 slow-a commit p0=2
09:14:36.168 slow-a p0 offset 2
09:14:36.171 slow-a commit p0=3
09:14:36.172 slow-a p0 offset 3 slow start
09:14:49.514 slow-b revoked []
09:14:49.521 slow-b assigned [0]
09:14:50.030 slow-b p0 offset 3
09:14:50.039 slow-b commit p0=4
09:14:50.544 slow-b p0 offset 4
09:14:50.548 slow-b commit p0=5
09:14:51.052 slow-b p0 offset 5
09:14:51.055 slow-b commit p0=6
09:14:51.560 slow-b p0 offset 6
09:14:51.563 slow-b commit p0=7
09:14:51.679 slow-a p0 offset 3
09:14:51.682 slow-a commit failed: KafkaError{code=UNKNOWN_MEMBER_ID,val=25,str="Commit failed: Broker: Unknown member"}
09:14:51.682 slow-a lost [0]
09:14:51.682 slow-a poll error: KafkaError{code=_MAX_POLL_EXCEEDED,val=-147,str="Application maximum poll interval (10000ms) exceeded by 356ms"}
09:14:52.068 slow-b p0 offset 7
09:14:52.072 slow-b commit p0=8
09:14:52.578 slow-b p0 offset 8
09:14:52.580 slow-b commit p0=9
09:14:52.580 slow-b revoked [0]
09:14:52.584 slow-a assigned [0]
09:14:52.584 slow-b assigned []
09:14:53.090 slow-a p0 offset 9
09:14:53.095 slow-a commit p0=10

%4|로 시작하는 첫 줄은 librdkafka가 남긴 경고예요. 1790813686.528은 유닉스 시간으로 09:14:46.528이고, offset 3 처리를 시작한 09:14:36.172에서 10.4초가 지난 시간입니다.

순서대로 따라가 볼게요.

1. 09:14:36.172: a가 offset 3을 받아 처리를 시작합니다. offset 0부터 2까지는 처리하고 p0=3까지 commit했습니다.
2. 09:14:46.528: 마지막 poll()에서 10초가 지나자 librdkafka가 Application maximum poll interval (10000ms) exceeded by 356ms를 남기고 그룹을 떠납니다(leaving group). a의 코드는 아직 offset 3을 처리하는 중이라 이 사실을 모릅니다.
3. 09:14:49.521: 대기하던 b가 파티션 0을 받고, commit된 3부터 읽어서 offset 3을 처리합니다. 이 순간부터 1.6초 동안 a와 b가 같은 메시지를 동시에 처리하고 있었습니다.
4. 09:14:51.679: a가 offset 3 처리를 마치고 commit하지만 UNKNOWN_MEMBER_ID로 실패합니다. a는 이미 그룹 구성원이 아니기 때문입니다. 이어서 on_lost가 호출되고, poll()은 _MAX_POLL_EXCEEDED 오류를 돌려줍니다.
5. 09:14:52.584: a가 그룹에 다시 들어오면서 리밸런스가 한 번 더 일어나고, 파티션 0은 다시 a에게 갔습니다.

4번에서 commit이 실패한 건 오히려 저장된 위치를 지키는 동작이에요. 그 시점에 b는 이미 p0=7까지 commit했어요. 만약 a의 p0=4 commit이 성공했다면, 그룹의 위치가 7에서 4로 돌아가서 offset 4부터 6까지를 또 처리했을 거예요.

정상 종료와 달리 a의 on_revoke는 호출되지 않았어요. 파티션을 이미 잃은 뒤라 on_lost가 대신 호출됐고, 이때는 commit해도 실패합니다. 그러니 on_revoke에서 commit하는 방법으로도 이 경우의 중복은 막을 수 없어요.

정리

범위: Kafka 4.3(apache/kafka:4.3.1) 단일 브로커, Python confluent-kafka 2.15.1, classic 프로토콜.

• on_revoke에서 commit하지 않은 그룹은 새 담당자가 commit된 400부터 읽어 82건을 두 번 처리했고, on_revoke에서 동기로 commit한 그룹은 중복이 0건이었어요. 파티션을 반납할 때 on_revoke에서 처리한 위치까지 동기로 commit하세요.
• offset 하나의 처리가 15초 걸리자 컨슈머는 max.poll.interval.ms 10초를 넘긴 10.4초 시점에 그룹을 떠났고, 1.6초 동안 두 컨슈머가 같은 메시지를 처리했어요. max.poll.interval.ms는 poll() 한 번에 받은 묶음의 최대 처리 시간보다 길게 잡아 주세요.
• 쫓겨난 컨슈머의 commit은 UNKNOWN_MEMBER_ID로 실패했고, on_revoke 대신 on_lost가 호출됐어요. 이 경우의 중복은 on_revoke commit으로 막을 수 없으니 처리 자체를 멱등하게 만들어 주세요.
• 이 글에 싣지 않은 추가 측정에서는, 리밸런스마다 모든 파티션을 반납하는 eager 할당(range) 그룹에 컨슈머가 하나 들어오자 파티션 6개 모두에서 중복 67건이 났고, 옮길 파티션만 반납하는 cooperative-sticky는 옮겨 간 파티션 2개에서만 24건이 났어요. eager 할당 대신 cooperative-sticky를 쓰고, 옮겨 가는 파티션을 위해 on_revoke commit은 유지하세요.
• 같은 추가 측정에서 group.instance.id 없는 컨슈머 하나를 재시작하자 다른 컨슈머가 파티션을 두 번 반납했고, group.instance.id(재시작해도 바뀌지 않는 인스턴스 id)를 준 static member는 띄운 지 0.14초 만에 리밸런스 없이 원래 파티션을 다시 받았어요. 재시작이 잦은 그룹에는 static membership을 쓰세요.