Kafka exactly-once는 정말 한 번만 처리할까?

Q: Kafka exactly-once는 정말 한 번만 처리할까?

트랜잭션을 쓰면 read_committed로 읽은 결과는 주문마다 하나였지만, 처리 도중 죽은 주문의 처리 코드는 두 번 실행됐고 출력 메시지도 파티션에 두 번 저장됐다.

이번 시간에는 Kafka exactly-once 트랜잭션이 무엇을 한 번으로 만들어 주고 무엇은 못 막는지를 KRaft 단일 노드 Kafka와 Python 처리 프로그램으로 프로세스를 죽여 보며 배워가는 시간입니다.

예를 들어서 orders 토픽에서 주문을 읽어 금액을 계산하고, 결과를 order-totals 토픽에 쓰는 서비스가 있다고 해볼게요. 두 번째 주문 O-2의 결과를 쓴 직후, 읽은 주문의 offset을 commit하기 전에 프로세스가 죽으면 어떻게 될까요?

Kafka의 exactly-once(EOS)는 결과 메시지를 쓰는 일과 입력 offset을 commit하는 일을 한 트랜잭션으로 묶어서, 둘 다 확정되거나 둘 다 확정되지 않게 해요. 같은 지점에서 죽는 상황을 트랜잭션 없이, 그리고 트랜잭션을 써서 재현하고 결과를 비교해 볼게요.

실습 준비

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

compose.yaml: KRaft 단일 노드 Kafka
processor.py: 주문을 읽어 금액을 계산해 출력 토픽에 쓰는 처리 프로그램
read.py: 토픽을 처음부터 격리 수준을 정해 읽는 스크립트

아래가 compose.yaml이고, 리스너는 두 개예요.

PLAINTEXT(kafka:9092): 컨테이너 안에서 Kafka 명령 도구가 써요.
HOST(localhost:9541): 호스트의 Python 프로그램이 써요.

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://:9541
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,HOST://localhost:9541
      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_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_HEAP_OPTS: -Xmx512m -Xms256m
    ports:
      - "9541:9541"

트랜잭션 상태는 __transaction_state라는 내부 토픽에 저장돼요. 이 토픽의 복제 계수 기본값은 3, min.insync.replicas 기본값은 2라서, 브로커가 한 대뿐이면 둘 다 1로 낮춰야 해요(KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR, KAFKA_TRANSACTION_STATE_LOG_MIN_ISR).

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

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

processor.py는 orders 파티션 0을 읽어 수량 x 단가로 금액을 계산하고, 결과를 출력 토픽에 써요. 인자는 이렇습니다.

첫째 인자: plain이면 트랜잭션 없이, tx면 트랜잭션으로 처리해요. 컨슈머 그룹은 각각 totals-plain, totals-tx예요.
둘째 인자: 결과를 쓸 출력 토픽이에요.
셋째 인자(선택): 이 주문을 처리할 때 결과를 쓴 뒤, offset을 확정하기 직전에 프로세스를 끝내요.

자동 commit은 끄고, 결과를 쓴 뒤에 offset을 commit해요. read.py는 토픽을 offset 0부터 읽어 offset과 값을 출력해요. 둘째 인자로 격리 수준(isolation.level)을 받습니다.

실습에서 반복해 쓸 셸 함수도 만들어 둘게요.

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

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

send() { kafka kafka-console-producer.sh --bootstrap-server kafka:9092 --topic "$1"; }

group() { kafka kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --group "$1"; }

txn() { kafka kafka-transactions.sh --bootstrap-server kafka:9092 describe --transactional-id "$1"; }

토픽을 만들고 주문 세 건을 넣을게요. 파티션은 모두 1개예요.

topic orders
topic order-totals
topic order-totals-tx
printf 'O-1 2x15000\nO-2 1x42000\nO-3 3x9000\n' | send orders

Created topic orders.
Created topic order-totals.
Created topic order-totals-tx.

실습 1: 트랜잭션 없이 offset commit 전에 죽으면

먼저 트랜잭션 없이 처리하다가, O-2의 결과를 쓴 직후 offset을 commit하기 전에 프로세스를 끝내 볼게요.

python processor.py plain order-totals O-2
group totals-plain

offset 0 O-1 처리, order-totals에 'O-1 total=30000' 씀
offset 1 O-2 처리, order-totals에 'O-2 total=42000' 씀
O-2: offset commit 전에 프로세스 종료

Consumer group 'totals-plain' has no active members.

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
totals-plain    orders          0          1               3               2               -               -               -

O-1은 처리와 commit을 모두 마쳐서 committed offset이 1이에요. O-2의 결과는 이미 order-totals에 들어갔지만, 그 사실은 offset에 반영되지 않았어요.

프로그램을 다시 시작하고 출력 토픽을 읽어 볼게요.

python processor.py plain order-totals
python read.py order-totals read_committed

offset 1 O-2 처리, order-totals에 'O-2 total=42000' 씀
offset 2 O-3 처리, order-totals에 'O-3 total=27000' 씀
[read_committed]
  offset 0: O-1 total=30000
  offset 1: O-2 total=42000
  offset 2: O-2 total=42000
  offset 3: O-3 total=27000

재시작한 프로그램은 offset 1부터 읽어서 O-2를 다시 처리했어요. 그래서 order-totals에는 O-2 total=42000이 offset 1과 2에 두 번 들어 있습니다. 트랜잭션 없이 쓴 메시지는 쓰는 순간 확정되기 때문에 read_committed로 읽어도 마찬가지예요.

결과를 쓰는 일과 offset을 commit하는 일은 서로 다른 요청이라, 둘 중 무엇을 먼저 하든 그 사이에서 죽는 순간이 있어요. commit 전에 죽으면 같은 메시지를 다시 처리하는 과정이 더 궁금하면 offset과 commit을 다룬 글을 참고하세요.

실습 2: 결과와 offset을 한 트랜잭션으로 묶기

Kafka에서 offset commit은 __consumer_offsets라는 내부 토픽에 메시지를 쓰는 일이에요. 그래서 프로듀서 트랜잭션 하나에 출력 토픽 쓰기와 __consumer_offsets 쓰기를 함께 넣을 수 있어요.

processor.py의 tx 모드는 주문 하나마다 다음 순서로 처리합니다.

transactional.id: 프로듀서에 붙이는 고정 이름이에요. 여기서는 order-totals-app이고, 프로세스가 재시작해도 같은 값을 써요.
init_transactions(): 시작할 때 한 번 불러요. 브로커의 트랜잭션 코디네이터에 이 transactional.id를 등록하고, 같은 id로 끝나지 않은 트랜잭션이 있으면 정리해요.
begin_transaction(): 트랜잭션을 시작해요.
produce(): 출력 토픽에 결과를 써요. 메시지는 파티션에 바로 저장되지만 아직 확정되지 않은 상태예요.
send_offsets_to_transaction(): 컨슈머의 다음 읽을 위치(position)와 그룹 정보를 트랜잭션에 넣어요. 이 offset도 트랜잭션이 커밋될 때 함께 확정돼요.
commit_transaction(): 출력 메시지와 offset을 한 번에 확정해요. 실패하면 abort_transaction()으로 둘 다 버립니다.

이번에도 같은 지점, 즉 O-2의 결과를 쓰고 offset을 트랜잭션에 넣은 뒤 commit_transaction()을 부르기 전에 프로세스를 끝내 볼게요. 그리고 kafka-transactions.sh로 이 transactional.id의 상태를 봅니다.

python processor.py tx order-totals-tx O-2
txn order-totals-app

offset 0 O-1 처리, order-totals-tx에 'O-1 total=30000' 씀
offset 1 O-2 처리, order-totals-tx에 'O-2 total=42000' 씀
O-2: offset commit 전에 프로세스 종료
CoordinatorId	TransactionalId 	ProducerId	ProducerEpoch	TransactionState	TransactionTimeoutMs	CurrentTransactionStartTimeMs	TransactionDurationMs	TopicPartitions
1            	order-totals-app	1         	0            	Ongoing         	60000               	1790819155673                	4013                 	order-totals-tx-0,__consumer_offsets-2

트랜잭션이 Ongoing 상태로 남아 있어요. TopicPartitions를 보면 출력 토픽 파티션(order-totals-tx-0)과 offset이 저장될 __consumer_offsets 파티션이 함께 들어 있어요.

같은 transactional.id로 프로그램을 다시 시작해 볼게요.

python processor.py tx order-totals-tx
txn order-totals-app
python read.py order-totals-tx read_committed
group totals-tx

%4|1790819160.412|GETPID|rdkafka#producer-2| [thrd:main]: Failed to acquire transactional PID from broker TxnCoordinator/1: Broker: Producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing: retrying
offset 1 O-2 처리, order-totals-tx에 'O-2 total=42000' 씀
offset 2 O-3 처리, order-totals-tx에 'O-3 total=27000' 씀
CoordinatorId	TransactionalId 	ProducerId	ProducerEpoch	TransactionState	TransactionTimeoutMs	CurrentTransactionStartTimeMs	TransactionDurationMs	TopicPartitions
1            	order-totals-app	1         	2            	CompleteCommit  	60000               	1790819160940                	9420
[read_committed]
  offset 0: O-1 total=30000
  offset 4: O-2 total=42000
  offset 6: O-3 total=27000

Consumer group 'totals-tx' has no active members.

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
totals-tx       orders          0          3               3               0               -               -               -

출력을 위에서부터 보면 이렇습니다.

새 프로세스의 init_transactions()가 처음에는 경고를 받고 재시도했어요. 코디네이터가 같은 id로 열려 있던 이전 트랜잭션을 중단하는 동안 기다린 거예요.
이전 트랜잭션의 offset commit이 확정되지 않았으니 committed offset은 여전히 1이에요. 새 프로세스는 offset 1의 O-2부터 다시 처리했어요.
read_committed로 읽은 결과에는 O-2 total=42000이 하나만 있어요. 컨슈머 그룹의 committed offset도 3으로 끝까지 진행했습니다.

실습 3: 처리 코드는 두 번 실행됐습니다

그런데 read_committed 결과의 offset이 0, 4, 6으로 띄엄띄엄이에요. 같은 토픽을 read_uncommitted로 읽어 볼게요.

python read.py order-totals-tx read_uncommitted

[read_uncommitted]
  offset 0: O-1 total=30000
  offset 2: O-2 total=42000
  offset 4: O-2 total=42000
  offset 6: O-3 total=27000

offset 2에 O-2 total=42000이 하나 더 있어요. 죽은 프로세스가 트랜잭션 안에서 쓴 메시지예요.

트랜잭션이 중단돼도 이미 파티션에 저장된 메시지는 지워지지 않아요. 대신 브로커가 이 메시지가 중단된 트랜잭션에 속한다는 기록을 남기고, read_committed 컨슈머는 그 기록을 보고 메시지를 걸러 내요. 빈 offset은 트랜잭션이 끝날 때 파티션에 쓰는 COMMIT·ABORT 표시(트랜잭션 마커)가 하나씩 차지한 자리예요.

그래서 트랜잭션으로 쓴 토픽을 읽는 쪽은 isolation.level을 확인해 주세요. 기본값이 클라이언트마다 달라요.

Java 컨슈머(kafka-console-consumer.sh 포함): read_uncommitted가 기본값이에요. 직접 read_committed로 바꾸지 않으면 중단된 트랜잭션의 메시지도 받아요.
librdkafka 기반 컨슈머(Python confluent-kafka 등): read_committed가 기본값이에요.

여기서 눈여겨볼 건 O-2를 처리하는 코드가 두 번 실행됐고, 출력 메시지도 파티션에 두 번 저장됐다는 점이에요. exactly-once에서 한 번으로 남는 건 확정된 결과뿐이에요. 처리 코드 안에서 로그를 남기거나 외부 API를 호출했다면 그 일은 두 번 일어납니다.

외부 시스템 쪽 중복은 이벤트 ID로 처리 기록을 남기거나 멱등 키를 쓰는 방식으로 막아야 해요. 그 방법이 궁금하면 at-least-once 처리를 다룬 글을 참고하세요.

정리

범위: Kafka 4.3.1 KRaft 단일 노드, Python confluent-kafka 2.15.1, 파티션 1개 토픽과 주문 세 건.

트랜잭션 없이 O-2 결과를 쓰고 offset commit 전에 죽자 재시작 뒤 O-2 total=42000이 offset 1과 2에 두 번 남았고, read_committed로 읽어도 같았습니다. 결과를 다른 토픽에 쓰는 서비스는 결과 쓰기와 offset commit을 한 트랜잭션으로 묶어 주세요.
같은 지점에서 죽어도 트랜잭션을 쓰면 재시작 때 이전 트랜잭션이 중단돼 read_committed로 읽은 O-2 결과는 하나였고, committed offset은 3까지 진행했습니다. offset은 send_offsets_to_transaction()으로 트랜잭션에 넣고 출력과 함께 commit_transaction()으로 확정하세요.
트랜잭션을 써도 O-2 처리 코드는 두 번 실행됐고 출력 메시지도 파티션에 두 번 저장됐습니다. 처리 코드 안의 로그와 외부 API 호출은 두 번 일어난다고 보고 따로 중복을 막아 주세요.
중단된 트랜잭션의 O-2는 read_uncommitted로 그대로 읽혔고, Java 컨슈머의 기본값은 read_uncommitted입니다. 트랜잭션으로 쓴 토픽을 읽는 쪽은 isolation.level=read_committed를 확인하세요.
확정된 메시지 세 개의 offset은 0, 4, 6으로 띄엄띄엄이었고, 빈 offset은 COMMIT과 ABORT 마커가 하나씩 차지했습니다. offset 차이로 메시지 수나 lag를 셀 때 마커와 중단된 메시지를 감안해 주세요.