응답이 사라지면 Kafka 멱등 프로듀서는 중복을 막을까?

Q: 응답이 사라지면 Kafka 멱등 프로듀서는 중복을 막을까?

멱등성을 끄면 응답만 사라진 이벤트가 offset 1과 2에 두 번 저장되고, 켜면 브로커가 같은 프로듀서 ID와 순번을 보고 한 번만 저장하지만 실패를 받은 애플리케이션이 다시 보낸 중복은 막지 못한다.

이번 시간에는 Kafka 프로듀서의 응답만 사라지는 상황을 직접 만들어서, 같은 메시지가 두 번 저장되는 과정을 offset으로 확인하고 멱등 프로듀서(enable.idempotence)가 그 중복을 어떻게 막는지, 그리고 어디까지 막지 못하는지 배워가는 시간입니다.

예를 들어서 결제 서비스가 E-42 결제 승인 이벤트를 Kafka에 보냈다고 해볼게요. 브로커는 이벤트를 파티션에 저장하고 응답을 돌려보냈는데, 그 응답이 프로듀서에 닿기 전에 사라졌어요.

프로듀서는 몇 초를 기다리다가 요청이 실패했다고 판단하고 같은 이벤트를 다시 보냅니다. 이때 토픽에는 E-42가 몇 번 남을까요?

실습 준비

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

compose.yaml: KRaft 단일 노드 Kafka와 리스너 세 개 정의
proxy.py: 지정한 순번의 Produce 응답만 버리는 테스트용 프록시(호스트 9532 → 컨테이너 9533)
producer.py: 프록시를 거쳐 send, app-retry 방식으로 이벤트를 보내는 프로듀서

compose.yaml의 리스너는 용도에 따라 나눠 두었어요.

PLAINTEXT(kafka:9092): 컨테이너 안에서 Kafka 명령 도구가 써요.
HOST(localhost:9531): 호스트에서 브로커에 바로 붙을 때 씁니다. 이 글에서는 쓰지 않아요.
FAULT(localhost:9532): 프로듀서 실습에서 써요. 호스트의 9532 포트에는 테스트용 프록시가 뜨고, 프록시가 컨테이너의 9533 포트로 요청을 넘깁니다.

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://:9531,FAULT://:9533
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,HOST://localhost:9531,FAULT://localhost:9532
      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:
      - "9531:9531"
      - "9533:9533"

컨테이너를 띄우고 Python 가상 환경을 만들어 주세요.

docker compose up -d
python3 -m venv venv && source venv/bin/activate
pip install confluent-kafka==2.15.1

응답이 사라지는 상황은 원하는 순간에 만들기 어려워서, 프로듀서와 브로커 사이에 작은 프록시를 두는 방식으로 가볼게요. 프록시는 지정한 순번의 Produce 요청이 지나가면 그 요청의 correlation id(요청과 응답을 짝짓는 번호)를 기억해 두고, 같은 correlation id를 단 응답이 브로커에서 돌아오면 프로듀서에 넘기지 않고 버립니다.

요청은 브로커에 그대로 들어가니까 메시지는 저장되고, 프로듀서만 결과를 모르게 되는 거죠. proxy.py <순번>이 이 프록시이고, 순번은 2나 1,2처럼 응답을 버릴 Produce 요청의 번호예요.

producer.py는 프록시(localhost:9532)를 거쳐 접속해요. 인자로 토픽, 멱등성 사용 여부, 보내는 방식을 받습니다.

send: E-41, E-42, E-43을 하나씩 보내고, 매번 결과를 받은 뒤 다음 이벤트를 보내요.
app-retry: E-42를 보내고, 실패 결과를 받으면 애플리케이션 코드가 produce()를 한 번 더 호출합니다.

socket.timeout.ms는 3초로 줄여 두었어요. 그래서 librdkafka가 응답 없는 요청을 3초 만에 실패로 보고 다시 보냅니다. 나머지 인자는 debug=eos처럼 설정으로 그대로 넘깁니다.

이제 실습에서 반복해 쓸 셸 함수를 만들어 놓을게요. dump는 kafka-dump-log.sh 출력에서 필요한 필드만 골라 한 줄로 보여 주는 함수이고, with_proxy <응답을 버릴 Produce 요청 순번> <명령...>은 프록시를 띄우고 명령을 실행한 뒤 프록시를 끕니다.

아래 블록은 파일로 만들지 말고 실습 디렉터리에서 연 터미널에 그대로 붙여 넣어 주세요. 이후 명령은 모두 이 터미널에서 실행하고, 터미널을 새로 열면 다시 붙여 넣으시면 됩니다.

kafka() { docker compose exec -T kafka /opt/kafka/bin/"$@"; }

topic() { kafka kafka-topics.sh --bootstrap-server kafka:9092 --create --topic "$1" --partitions 1; }

read_all() {
  kafka kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic "$1" --from-beginning --timeout-ms 3000 \
    --formatter-property print.offset=true 2> /dev/null
}

dump() {
  kafka kafka-dump-log.sh --print-data-log --files "/tmp/kafka-logs/$1-0/00000000000000000000.log" |
    awk '/^baseOffset/ { for (i = 1; i < NF; i += 2) f[$i] = $(i + 1)
           printf "baseOffset: %s producerId: %s producerEpoch: %s baseSequence: %s CreateTime: %s crc: %s",
             f["baseOffset:"], f["producerId:"], f["producerEpoch:"], f["baseSequence:"], f["CreateTime:"], f["crc:"] }
         /payload:/ { print " payload: " $NF }'
}

with_proxy() {
  python proxy.py "$1" & local px=$!
  sleep 1
  "${@:2}"
  kill $px
}

실습 1: 멱등성을 끄면 같은 이벤트가 두 번 저장됩니다

먼저 파티션이 1개인 pay-off 토픽을 만들고, 멱등성을 끈 프로듀서로 이벤트 세 건을 보내 볼게요. 프록시는 두 번째 Produce 요청, 즉 E-42의 응답을 버립니다.

topic pay-off
with_proxy 2 python producer.py pay-off false send
read_all pay-off

Created topic pay-off.
 0.0s E-41 전송 결과: offset=0
proxy: Produce 응답 correlation_id=3 버림
%5|1790817116.067|REQTMOUT|rdkafka#producer-1| [thrd:localhost:9532/1]: localhost:9532/1: Timed out ProduceRequest in flight (after 3038ms, timeout #0)
%4|1790817116.067|REQTMOUT|rdkafka#producer-1| [thrd:localhost:9532/1]: localhost:9532/1: Timed out 1 in-flight, 0 retry-queued, 0 out-queue, 0 partially-sent requests
 3.2s E-42 전송 결과: offset=2
 3.2s E-43 전송 결과: offset=3
Offset:0	E-41
Offset:1	E-42
Offset:2	E-42
Offset:3	E-43

일어난 일을 시간순으로 보면 이래요.

E-41은 바로 offset 0에 저장됐습니다.
E-42 요청은 브로커에 들어가 offset 1에 저장됐지만, 프록시가 응답을 버렸어요.
3초가 지나자 librdkafka가 요청을 타임아웃으로 처리하고(REQTMOUT 로그) 같은 메시지를 다시 보냈습니다. 브로커는 이 요청을 새 메시지로 받아 offset 2에 저장했어요.
프로듀서가 받은 결과는 E-42 전송 결과: offset=2 하나뿐입니다. 애플리케이션 입장에서는 E-42를 한 번 보내 한 번 성공한 거예요.

kafka-dump-log.sh로 파티션의 로그 세그먼트 파일을 직접 읽어 볼게요. 프로듀서는 메시지를 배치로 묶어 보내고, 브로커는 받은 배치를 그대로 세그먼트에 붙입니다.

dump pay-off

baseOffset: 0 producerId: -1 producerEpoch: -1 baseSequence: -1 CreateTime: 1790817113002 crc: 146135169 payload: E-41
baseOffset: 1 producerId: -1 producerEpoch: -1 baseSequence: -1 CreateTime: 1790817113029 crc: 46838451 payload: E-42
baseOffset: 2 producerId: -1 producerEpoch: -1 baseSequence: -1 CreateTime: 1790817113029 crc: 46838451 payload: E-42
baseOffset: 3 producerId: -1 producerEpoch: -1 baseSequence: -1 CreateTime: 1790817116158 crc: 1877229245 payload: E-43

producerId: 배치를 보낸 프로듀서의 ID(PID)예요. 멱등성을 쓰지 않으면 -1입니다.
producerEpoch: 같은 PID 안에서 세대를 구분하는 번호예요.
baseSequence: 배치의 첫 메시지 순번이에요. 멱등성을 쓰지 않으면 -1입니다.
CreateTime, crc: 프로듀서가 메시지를 만든 시간과 배치 내용으로 계산한 체크섬이에요.

offset 1과 2의 CreateTime과 crc가 같아요. 클라이언트가 처음 만든 배치를 바이트 그대로 다시 보냈기 때문이죠. 그런데 producerId와 baseSequence가 -1이라서, 브로커에는 두 배치가 같은 것인지 알아볼 단서가 없어요.

실습 2: 멱등성을 켜면 한 번만 저장됩니다

이번에는 같은 실험을 enable.idempotence=true로 다시 해 볼게요.

librdkafka 디버그 로그(debug=msg,eos)도 켜서 프로듀서가 보낸 배치에 무엇이 붙는지 같이 봅니다. grep으로 필요한 줄만 남기고, sed로 디버그 로그 줄 앞의 시간과 스레드 정보는 잘라 냈어요.

topic pay-on
with_proxy 2 python producer.py pay-on true send debug=msg,eos 2>&1 \
  | grep -E 'Acquired PID|PRODUCE|MSGSET|전송 결과|proxy' | sed -E 's/^%7\|[0-9.]+\|[A-Z]+\|rdkafka#producer-1\| \[thrd:[^]]*\]: //'
read_all pay-on

Created topic pay-on.
Acquired PID{Id:0,Epoch:0}
localhost:9532/1: pay-on [0]: Produce MessageSet with 1 message(s) (72 bytes, ApiVersion 10, MsgVersion 2, MsgId 1, BaseSeq 0, PID{Id:0,Epoch:0}, uncompressed)
localhost:9532/1: pay-on [0]: MessageSet with 1 message(s) (MsgId 1, BaseSeq 0) delivered
 1.0s E-41 전송 결과: offset=0
localhost:9532/1: pay-on [0]: Produce MessageSet with 1 message(s) (72 bytes, ApiVersion 10, MsgVersion 2, MsgId 2, BaseSeq 1, PID{Id:0,Epoch:0}, uncompressed)
proxy: Produce 응답 correlation_id=5 버림
localhost:9532/1: pay-on [0]: MessageSet with 1 message(s) (MsgId 2, BaseSeq 1) encountered error: Local: Timed out (actions Retry,MsgPossiblyPersisted)
localhost:9532/1: pay-on [0]: Produce MessageSet with 1 message(s) (72 bytes, ApiVersion 10, MsgVersion 2, MsgId 2, BaseSeq 1, PID{Id:0,Epoch:0}, uncompressed)
localhost:9532/1: pay-on [0]: MessageSet with 1 message(s) (MsgId 2, BaseSeq 1) delivered
 4.2s E-42 전송 결과: offset=1
localhost:9532/1: pay-on [0]: Produce MessageSet with 1 message(s) (72 bytes, ApiVersion 10, MsgVersion 2, MsgId 3, BaseSeq 2, PID{Id:0,Epoch:0}, uncompressed)
localhost:9532/1: pay-on [0]: MessageSet with 1 message(s) (MsgId 3, BaseSeq 2) delivered
 4.2s E-43 전송 결과: offset=2
Offset:0	E-41
Offset:1	E-42
Offset:2	E-43

출력을 보면 토픽에는 E-42가 offset 1에 하나만 있어요. 로그를 위에서부터 읽으면 과정이 보입니다.

프로듀서는 메시지를 보내기 전에 브로커에서 PID를 받았어요(Acquired PID{Id:0,Epoch:0}).
배치마다 PID와 순번이 붙었습니다. E-41은 BaseSeq 0, E-42는 BaseSeq 1이에요. 순번은 PID와 파티션 조합마다 0부터 셉니다.
E-42의 응답이 사라지자 librdkafka는 이 요청을 MsgPossiblyPersisted, 즉 저장됐을 수도 있는 메시지로 분류하고 재시도했어요.
재시도한 배치에도 처음과 같은 BaseSeq 1, PID{Id:0,Epoch:0}이 붙었습니다.
브로커는 이 PID에서 순번 1을 이미 저장했다는 것을 알아보고, 다시 저장하지 않은 채 처음 저장한 offset 1로 성공 응답을 보냈어요. 그래서 전송 결과가 offset=1입니다.

브로커는 PID마다, 파티션마다 최근에 저장한 배치 5개의 순번을 기억해요. 같은 PID에서 이미 저장한 순번의 배치가 다시 오면 중복으로 보고 저장하지 않습니다.

실습 3: 애플리케이션이 다시 send하면 막지 못합니다

멱등성은 클라이언트가 같은 배치를 다시 보낼 때 PID와 순번을 그대로 붙이는 방식으로 동작해요. 그런데 애플리케이션이 produce()를 다시 부르면 어떻게 될까요? 클라이언트 입장에서는 새 메시지이므로 새 순번이 붙습니다.

애플리케이션이 다시 보내는 흔한 경우는 전송 결과로 실패를 받았을 때예요. 클라이언트는 재시도를 계속하다가 전송 제한 시간이 지나면 메시지를 실패로 돌려줍니다. 이 제한 시간은 Java 클라이언트의 delivery.timeout.ms(기본값 120000, 2분), librdkafka의 message.timeout.ms(기본값 300000, 5분)예요.

실습에서는 message.timeout.ms를 5초로 줄이고, 프록시가 첫 요청과 재시도 요청의 응답을 모두 버리게 할게요. app-retry 모드는 실패를 받으면 E-42를 한 번 더 보냅니다.

topic pay-app
with_proxy 1,2 python producer.py pay-app true app-retry message.timeout.ms=5000 debug=eos 2>&1 \
  | grep -E 'NEWPID|전송 결과|proxy' | sed -E 's/^%7\|[0-9.]+\|[A-Z]+\|rdkafka#producer-1\| \[thrd:[^]]*\]: //'
read_all pay-app
dump pay-app

Created topic pay-app.
pay-app [0] changed PID{Invalid} -> PID{Id:1,Epoch:0} with base MsgId 1
proxy: Produce 응답 correlation_id=3 버림
proxy: Produce 응답 correlation_id=5 버림
 5.1s E-42 전송 결과: KafkaError{code=_MSG_TIMED_OUT,val=-192,str="Local: Message timed out"}
pay-app [0] changed PID{Id:1,Epoch:0} -> PID{Id:1,Epoch:1} with base MsgId 2
 5.1s E-42 전송 결과: offset=1
Offset:0	E-42
Offset:1	E-42
baseOffset: 0 producerId: 1 producerEpoch: 0 baseSequence: 0 CreateTime: 1790817170983 crc: 746468498 payload: E-42
baseOffset: 1 producerId: 1 producerEpoch: 1 baseSequence: 0 CreateTime: 1790817176038 crc: 265925614 payload: E-42

출력을 보면 멱등성을 켰는데도 E-42가 두 번 저장됐어요.

첫 전송은 offset 0에 저장됐지만 응답이 사라졌고, 클라이언트의 자동 재시도도 응답을 받지 못했습니다. 자동 재시도는 같은 순번이라 브로커가 다시 저장하지 않았어요.
5초가 지나자 클라이언트는 _MSG_TIMED_OUT으로 실패를 알렸습니다. 실제로는 offset 0에 저장된 메시지예요.
librdkafka는 결과를 모르는 메시지를 실패로 돌려준 뒤 epoch를 0에서 1로 올렸어요(PID{Id:1,Epoch:0} -> PID{Id:1,Epoch:1}). 실패 처리한 메시지 때문에 순번이 어긋나지 않도록 순번을 새로 세기 위해서입니다.
애플리케이션이 다시 보낸 E-42는 epoch 1, 순번 0을 달고 offset 1에 저장됐어요.

브로커가 보기에 offset 0은 (PID 1, epoch 0, 순번 0), offset 1은 (PID 1, epoch 1, 순번 0)인 서로 다른 배치입니다. 그러니 전송 실패를 받은 메시지도 이미 저장됐을 수 있다고 생각해 주세요.

정리하면 멱등 프로듀서가 막는 건 한 프로듀서 인스턴스 안에서 클라이언트 라이브러리가 자동으로 재시도해서 생기는 중복이에요. 애플리케이션이 send()나 produce()를 다시 부른 경우, 프로세스가 재시작해 새 PID로 보낸 경우, 서로 다른 프로듀서가 같은 이벤트를 보낸 경우는 막지 못합니다.

이런 중복이 생길 수 있다면 같은 이벤트에는 같은 이벤트 ID를 넣어 보내 주세요. 컨슈머 쪽에서 이벤트 ID로 걸러 내는 방법이 더 궁금하면 at-least-once 처리를 다룬 글을 참고하세요.

정리

범위: Kafka 4.3(apache/kafka:4.3.1) KRaft 단일 노드, Python confluent-kafka 2.15.1, 파티션 1개, 응답을 버리는 프록시로 만든 유실.

멱등성을 끈 프로듀서는 응답만 사라진 E-42를 3초 뒤 다시 보내 offset 1과 2에 두 번 저장했고, 애플리케이션이 받은 결과는 성공 한 번뿐이었습니다. 응답이 없다고 해서 저장되지 않았다고 보지 마세요.
멱등성을 켜자 재시도한 배치가 같은 PID{Id:0,Epoch:0}, BaseSeq 1을 달고 갔고, 브로커는 다시 저장하지 않은 채 offset 1로 응답했습니다. librdkafka 기반 클라이언트는 기본값이 꺼져 있으니 enable.idempotence=true를 직접 켜 주세요.
message.timeout.ms 5초에 _MSG_TIMED_OUT을 받은 E-42는 이미 offset 0에 저장돼 있었고, 애플리케이션이 다시 보낸 E-42는 epoch 1, 순번 0으로 offset 1에 또 저장됐습니다. 전송 실패를 받은 이벤트를 다시 보내는 코드는 중복이 생긴다고 보고 설계하세요.
이 글에 싣지 않은 추가 측정에서는 응답 전에 죽은 프로세스는 PID 2로, 재시작한 프로세스는 PID 3으로 E-42를 순번 0에 보내 두 번 저장됐습니다. 재시작 뒤 다시 보낼 수 있는 이벤트에는 같은 이벤트 ID를 넣어 컨슈머가 걸러 내게 해 주세요.