개요

카프카 스트림즈에서 Exactly-once Sementic(EOS)은 processing.guarantee로 설정이 가능하다.

기본값은 "at_least_once" 이며, "exactly_once"(EOS version 1), "exactly_once_v2"(EOS version 2) 를 선택할 수 있다. 

version 2는 version 1에 비해 개선된 버전이고, version 1은 최신버전에선 deprecated 된 상황이다.

따라서, 카프카 2.5 이상의 버전에서 EOS 가 필요하다면 version 2를 선택하는 것이 유일한 방법이다.

이 문서에서는 EOS version 1, version 2 에 대해 살펴봄으로써 다음의 목표를 달성하고자 한다.

• 카프카 스트림즈가 제공하는 Exactly-once의 동작 원리를 파악한다.
• EOS의 각 버전의 효과와 한계를 파악한다.
• 한계 파악을 통해, 라이브 환경에서 발생 가능한 문제를 추론하고 이에 대한 대응책을 모색한다.

EOS version 1(Kafka 0.11)

Transactions in Apache Kafka | Confluent
Enabling Exactly-Once in Kafka Streams | Confluent

Exactly-once가 깨지는 원인

1) 중복 쓰기

• 프로듀서가 브로커로부터 ack 응답을 받지 못하는 경우, 중복 쓰기가 발생할 수 있다.
• 해결: 멱등성 프로듀서로 해결한다.

2) 중복 컨슘

• sink 토픽으로 발행 후, source 토픽으로 offset을 커밋하기 직전 인스턴스가 다운 되는 경우, 중복 컨슘하여 두 번 처리될 수 있다.
• 해결: 트랜잭션 프로듀서/컨슈머로 해결한다.

3) zombie instances 문제

• 분산 환경에서 인스턴스가 나머지 시스템으로부터 연결이 일시적으로 끊긴 경우, 새 인스턴스가 기존 인스턴스를 교체하기 위해 시작될 수 있다.
• 이 경우, 같은 소스와 싱크를 바라보고 동일한 처리를 하는 인스턴스가 2개 이상 발생할 수 있다. 이러한 경우를 zombie instances 라고 부른다.
• 트랜잭션 프로듀서가 적용된 상황에선, 동일한 txId 를 갖되, 트랜잭션으로 발행하는 레코드에 있는 epoch 값이 브로커가 가지고 있는 값보다 미만인 프로듀서 인스턴스라고 볼 수 있겠다.
• 해결: 트랜잭션 프로듀서/컨슈머로 해결한다.

카프카의 EOS 접근 방식

스트림 프로세싱 애플리케이션에서의 Exactly-once 는 다음과 같이 정의할 수 있다.

• 하나의 입력 레코드로 상태가 정확히 한 번 업데이트 되고, 출력 레코드가 정확히 한 번 발행되는 것.

카프카 스트림즈 애플리케이션에서의 단대단 일관성이란 다음 조건을 만족시키는 것으로 해석될 수 있다.

1. change_log 토픽으로 레코드를 발행한다.(상태를 업데이트한다.)
2. sink 토픽으로 레코드를 발행한다.(처리 결과를 방출한다.)
3. __consumer_offsets 토픽으로 레코드를 발행한다.(처리 완료한 소스 레코드에 대해 커밋한다.)

위 세 개의 레코드 발행 동작이 트랜잭션으로 묶여서 원자적으로 동작한다면, 카프카 스트림즈 애플리케이션은 일관성을 갖출 수 있게 된다.

• 즉, 다음 명제가 성립할 수 있다.
  • "change_log 토픽의 오프셋이 증가" <=> "sink 토픽의 오프셋이 증가" <=> "__consumer_offsets의 offset이 증가"
• 세 개의 토픽의 offset 증가가 동기화 됨으로써, 커밋된 소스 레코드가 정확히 한 번 처리되어 상태를 정확히 한 번 업데이트하고, 정확히 한 번 발행되었다는 것이 보장되는 것이다.
• 만약, 인스턴스가 다운되어 다시 복구될 때, 이전 트랜잭션까지 상태 저장소가 복구되고, 이전의 커밋되었던 소스서부터 다시 처리를 시작하게 될 것이다.

한 편, EOS version 1 에서는 다음의 전략으로, 위에서 살펴본 Exactly-once 문제를 해결했다.

1. 중복 쓰기: 멱등성 프로듀서를 사용하여 해결했다.
2. 중복 컨슘: 트랜잭션 프로듀서를 통해, sink 토픽으로 출력과 source 토픽 커밋을 트랜잭션으로 묶음으로써 해결했다.
3. zombie instances: zombie fencing(아래에서 설명)으로 해결했다.

트랜잭션 동작 상세 메커니즘(KIP-98)

KIP-98 - Exactly Once Delivery and Transactional Messaging

Apache Kafka’s Exactly-Once Semantics Are Now Easier & More Robust (confluent.io)

브로커의 TransactionCoordinator 모듈: 각 트랜잭션를 관리

• 하나의 txId에 하나의 프로듀서를 할당하도록 관리한다.
• txId에 producer ID 와 producer epoch를 할당하는 방법으로 구현

producer ID(pid)

• txId와 함께 프로듀서가 브로커에게 초기화 요청(InitProducerId API, step2)을 보냄
• 브로커는 pid를 생성해서 txId에 할당하고 프로듀서에게 응답
• 이미 txId에 pid가 할당되어 있는 값이면, 해당 값을 프로듀서에게 응답하고, epoch 값을 증가 시킴.

producer epoch

• pid와 연결되며, 해당 pid에 새 프로듀서 인스턴스가 초기화될 때마다 증가함
• 브로커는 pid와 현재 epoch가 있는 프로듀서에서만 데이터를 쓰거나 커밋할 수 있도록 허용됨.
• 로그에 있는 각 레코드는 pid 와 epoch 값을 갖고 있음. 브로커가 시작되면, 각 파티션의 로그를 읽으며, pid를 현재 epoch에 매핑하여 캐시에 저장.

zombie fencing

• 프로듀서 인스턴스의 epoch 값보다 미만인 경우 fencing 하는 방법.
• ProduceRequest 엔 pid, epoch, sid가 함께 담겨 있음
• 프로듀서의 발행 요청(ProduceRequest, step4.2)의 pid가 브로커의 캐시에 있는지, 있다면 현재 epoch 값 이상인지 점검함. 미만이라면 fencing 함

한계

한계 1: 에러 핸들링 문제

프로듀서 에러 핸들링 문제

• 프로듀서가 EOS를 준수하며 레코드를 안전하게 보낼 수 있는지 결정할 수 있는 로직이 프로듀서와 브로커 둘 다 가지고 있다.
  • pid, epoch 값이 브로커와 프로듀서가 둘 다 가지고 있음
  • 프로듀서가 보내는 메시지의 저 값들이 브로커의 현재 값의 조건을 충족하는지 따짐
    -프로듀서나 브로커가 위와 같은 결정을 내리지 못하는 상태가 되면, 프로듀서는 fatal 에러 상태가 된다.
• 프로듀서의 fatal 에러 상태를 해결할 수 있는 유일한 방법은 프로듀서를 닫고 새로 생성하는 것이었다.
• 이런 경우, 카프카 스트림즈 애플리케이션에선 태스크의 이관과 리밸런싱이 수반되고 처리율이 낮아진다.

트랜잭션 프로듀서의 fatal error1: UNKNOWN_PRODUCER_ID error

• 카프카 2.5 이전에는 로그에 더 이상 레코드를 갖고 있지 않은 pid를 캐시에서 삭제했다.
  • 예를 들어, 트랜잭션 사이의 모든 레코드들의 retention 시간이 모두 지나가도 애플리케이션이 레코드를 발행하지 않는 경우, 위와 같이 pid가 삭제될 수 있다.
• 이처럼 pid가 삭제된 후, 해당 pid의 프로듀서가 레코드 발행 시도를 하는 경우, 브로커는 UNKNOWN_PRODUCER_ID 에러를 프로듀서에게 반환하고, 그 프로듀서는 fatal error 상태에 진입하게 된다.
  • 프로듀서가 가지고 있는 pid, epoch는 의미상 유효하지만, 브로커의 캐시에서 삭제됨으로써 에러가 발생하게 된 것
• 동작 중인 브로커에게 txId, pid 를 할당하기 위해선 프로듀서의 초기화 요청(InitProducerId API)을 호출하는 방법뿐이다.
  • 그런데, 이 요청은 오직 프로듀서의 생애 주기의 시작 단계에서만 호출 가능하다.
  • 따라서, 이 프로듀서를 복구하는 유일한 방법은 새로운 프로듀서를 생성하고 새 pid로 초기화하는 것뿐이다.

멱등성 프로듀서의 fatal error2: sid(sequence ID) 할당 에러

• 프로듀서가 sid를 할당할 수 없을 때, 발생하는 에러
• 멱등적 프로듀서의 발행 요청이 retriable 에러로 실패한다면, 발행 성공하거나 delivery.timeout.ms 까지 발행 재시도함
• delivery.timeout.ms 가 지나면 프로듀서에 의해 실패한 레코드가 만료 처리됨
• 만료된 레코드에 대해서 프로듀서는 발행이 성공/실패 했는지 확신할 수 없음
• 이런 경우, 프로듀서는 다음 레코드에 어떤 sid 할당할지 알 수 없고, 처리를 지속할 수 없게 됨
• 이를 sid 초기화로 해결할 수 있겠지만, 이는 새로운 프로듀서 생성 말고는 달성 불가능.

한계 2: 스케일링 문제

• 이전에는 컨슈머에게 파티션이 정적으로 할당된 경우에만 트랜잭션 프로듀서가 동작했다.
  • (예시) 트랜잭션 프로듀서 P1이 트랜잭션을 완료하기 전에 충돌이 일어나서, 입력 파티션 tp-1가 다른 트랜잭션 프로듀서 P2에게 할당 되었다고 하자.
  • P2는 tp-1 의 실패 off-set 부터 처리한 후, 커밋을 할 것이다.
  • 이와 동시에 P1이 트랜잭션을 재시도 하여 tp-1에 대해 커밋을 할 것이다.
  • 결국 같은 파티션 tp-1 에 대해 중복 처리가 되는 것이다.
• 그러나, 카프카에선 컨슈머 그룹은 동적으로 스케일링되면서 파티션 역시 동적으로 재할당 된다.
• 정적 할당을 충족시키기 위해선, 각 입력 파티션이 별도의 프로듀서 인스턴스를 사용해야했다.
  • 위에서 살펴본 (예시)에서와 같은 일이 일어나지 않으려면, 하나의 트랜잭션 프로듀서에 입력 토픽 파티션이 정적으로 할당되어야 한다는 것을 의미한다.
• 이로 인해, 카프카 브로커와 많은 수의 클라이언트 연결이 필요해지는데, 이는 브로커의 안정성에 악영향을 미치며 리소스를 낭비시킨다.

EOS version 2(Kafka 2.5)

version 1에 비해 향상된 버전의 EOS이다.

version 1은 프로듀서가 파티션 별로 생성되지만, version 2는 프로듀서가 스레드 별로 생성된다.

멱등성/프로듀서 프로듀서 신뢰성 향상(KIP-360)

카프카 2.5 버전에서는 프로듀서의 에러 핸들링 문제를 해결하고자 EOS version 1 와 같은 문제가 발견되면,
프로듀서를 새로운 인스턴스로 초기화 할 필요 없이 자동으로 복구하는 메커니즘을 추가했다.

safe epoch bumping(InitProcessorId API)

• InitProducerId API 가 txId 뿐 아니라, optional 값으로 pid, epoch 값을 함께 받아서 브로커로 요청하게 된다.
• 이 값들이 함께 있는 경우, 브로커는 현재 txId에 할당된 pid 와 epoch 값과 비교한다.
• 일치하는 경우, 해당 txId로 다른 생산자가 초기화되지 않은 것이므로, 이는 기존 프로듀서가 계속 처리해도 안전한 것이다. 브로커는 epoch 값을 증가시키고 기존 프로듀서에게 반환한다.
  
  If the existing producer’s ID and epoch match the transaction metadata, the transaction coordinator bumps the epoch and the producer continues processing.
• 기존 프로듀서가 다시 복구되기 전에 새로운 프로듀서가 초기화된 경우, epoch를 0으로 리셋하고, 새로운 pid를 발급한다. 기존 프로듀서는 fenced 된다.
  
  On the other hand, if a new producer has been initialized before the existing producer can re-initialize, the existing one will be fenced.
• epoch 값이 증가하거나 리셋되어 변경되는 경우(bumped), sid 를 0으로 리셋시킴으로써 앞선 fatal error1, fatal error2 발생시에도 프로듀서가 계속 처리 가능하게 한다.

프로듀서 자동 복구 메커니즘

• 프로듀서에서 fatal errors 가 발생하고, 새로운 버전의 InitProducerId 를 지원한다면, fatal error state 가 아닌, abortable error state로 전환됨.
• 애플리케이션이 트랜잭션을 중단(abort)하면, 프로듀서는 내부적으로 InitProducerId 를 호출한다. 이에 따라 epoch 가 증가하고, 계속해서 처리할 수 있게 된다.
  • 의문: UNKNOWN_PRODUCER_ID 에러의 경우, 브로커에 이미 해당 pid 가 사라진 뒤인데, 새로운 InitProducerId가 어떻게 이를 해결할 수 있나? 프로듀서가 pid와 epoch값을 가지고, 트랜잭션 실패 후 재초기화 요청을 보낸다고 해도, 비교할 수 있는 pid가 브로커에겐 없을텐데...?

프로듀서 자동 복구 메커니즘이 fatal error1 을 해결할 수 있는 이유(검증 필요)

• 생애 주기의 시작 단계에서만 InitProducerId 를 호출하는 것이 아닌, UNKNOWN_PRODUCER_ID 를 반환 받아도 호출할 수 있게 됨. → 새로운 프로듀서 생성 필요 없음
• 의문: UNKNOWN_PRODUCER_ID 에러의 경우, 브로커에 이미 해당 pid 가 사라진 뒤인데, 새로운 InitProducerId가 어떻게 이를 해결할 수 있나? 프로듀서가 pid와 epoch값을 가지고, 트랜잭션 실패 후 재초기화 요청을 보낸다고 해도, 비교할 수 있는 pid가 브로커에겐 없을텐데...?

프로듀서 자동 복구 메커니즘이 fatal error2 을 해결할 수 있는 이유(검증 필요)

• 인스턴스를 새로 생성할 필요없이 재초기화가 가능해짐
• 재초기화 후, 자동으로 sid가 0으로 리셋되므로 해결 됨

의문: 이전 버전의 InitProcessorAPI를 재초기화 단계에서 호출하는 것은 어딘가 불완전한 방법이어서 새로운 버전을 개발한 것인가? 만약 호출하는 것이 가능하다면, 이전 버전의 API론  fatal error를 해결할 수 없는건가?

EOS를 위한 프로듀서 확장성(KIP-447)

브로커의 GroupCoordinator 가 트랜잭션 프로듀서 간의 파티션 재할당을 모니터하게 됨

= 본래 GroupCoordinator 는 커밋된 오프셋을 유지하고, 컨슈머 그룹의 리밸런싱을 중재하도록 지정된 브로커이다.

• 트랜잭션으로 인해 제출되는 오프셋 커밋이 컨슈머 그룹을 통해 흐르기 때문에, 이미 GroupCoordinator 는 컨슈머 그룹의 프로세싱을 알고 있다. 
• 트랜잭션이 완료되지 않은 상태에서 일부 오프셋이 GroupCoordinator에게 커밋되면 이 오프셋을 보류 중인 오프셋으로 취급한다.
  이 보류 중인 오프셋은 완료되지 않은 트랜잭션의 표시로 활용될 수 있다.

오프셋 패치 프로토콜 보호 기능 강화

• 컨슈머 그룹의 리밸런싱이 완료되었을 때, 컨슈머 그룹의 파티션이 보류 중인 오프셋을 갖고 있는 경우, 관련 트랜잭션이 완료될 때까지 오프셋 패치(가져오기) 호출이 차단됨.
• 이전 버전에서는 패치 호출이 차단되지 않고 지난 오프셋을 반환했으며, 애플리케이션이 즉시 프로세싱을 계속하도록 했었음.
  
  P2가 tp-1에 대해 트랜잭션을 처리하다가 연결이 끊긴 경우, GroupCoordinator는 tp-1의 트랜잭션 완료 전 제출된 오프셋에 대해 보류 표시를 하고, tp-1을 P1에게 할당한다.
  
  GroupCoordinator가 tp-1의 보류 오프셋을 타임아웃 처리할 때까지, P1은 tp-1에 대해 오프셋 패치 요청을 재시도한다.

좀비 프로듀서의 엣지 케이스를 대응하는 로직이 추가되었다.

• 컨슈머 그룹의 메타데이터 중 group generation 이란 것이 존재한다. 이 값은 리밸런싱 후에, bumped 된다.
• 컨슈머 그룹과 트랜잭션 시멘틱을 통합함으로써, zombie 프로듀서가 generation이 만료되어도, 유효한 epoch로 커밋할 수 있게 되었다.
• 프로듀서의 API인 sendOffsetsToTransaction 은 컨슈머 메타데이터를 인수로 포함하면서 group generation 에 따라 fence 될 수 있게 되었다.

한계

참고 사항

카프카 스트림즈는 트랜잭션 프로듀서/컨슈머와 멱등성 프로듀서를 사용한다.

• 즉, 스트림즈 설정 값인 processing.guarantee = "exactly_once" || "exactly_once_v2" 로 설정하면 다음의 설정 값이 자동으로 설정된다.
• 프로듀서가 transaction_id 를 사용하고, 컨슈머는 isolation.level = "read_committed" 로  동작한다.
• 프로듀서가 enable.idempotence=true 로 동작한다.

이로 인해, 처리율은 떨어질 수 있음을 인지해야 한다.

Flink 와의 비교

아래는 gpt o3 로 확인한 내용이므로 검증이 필요하다.

Flink 에서 소스와 싱크를 모두 kafka 를 사용하였을 때, 스냅샷 사이의 모든 소스 offset 커밋과 싱크 producing 을 하나의 trx 로 묶어서 exactly-once 를 보장한다.

Kafka Streams 의 경우도, 마찬가지로 매 컨슘한 레코드 하나마다 trx 를 유지하는 것이 아니라, 몇 개의 레코드를 묶어서 하나의 trx 로 처리한다고 한다.

참고 자료

Transactions in Apache Kafka | Confluent
Enabling Exactly-Once in Kafka Streams | Confluent
streams-developer-guide-processing-guarantee | Apache Kafka
Exactly Once Semantics :: Spring Kafka
KIP-98 - Exactly Once Delivery and Transactional Messaging
Apache Kafka’s Exactly-Once Semantics Are Now Easier & More Robust (confluent.io)