4. Kafka Offsets and Exactly-Once

Kafka 쪽에서 레코드가 어떻게 쓰이는지와 Flink가 offset을 어떻게 믿는지는 Sulco 발표로 따로 정리했다. 이 장에서는 그중 Flink 쪽, 즉 checkpoint와 Kafka가 맞물리는 방식과 exactly-once 설정을 다룬다.

4.1 Offsets Live in the Checkpoint, Not in Kafka

처음 보면 가장 헷갈리는 부분이다. 일반적인 Kafka consumer는 어디까지 읽었는지를 Kafka에 commit하고, 재시작하면 commit된 offset부터 다시 읽는다.

Flink의 KafkaSource는 다르다. 읽은 위치를 checkpoint 안에 operator state로 저장하고, 복구할 때도 checkpoint에 있는 offset을 쓴다. Kafka에 commit하는 offset은 checkpoint가 끝난 뒤에 뒤따라 갱신될 뿐이다. 공식 문서도 이 값은 진행 상황을 보여 주는 용도이고 복구에는 쓰지 않는다고 설명한다.

checkpoint 하나가 도는 동안 source와 sink가 하는 일을 순서대로 따라가 보면 이렇다.

sequenceDiagram
  participant JM as JobManager
  participant SRC as KafkaSource
  participant SNK as KafkaSink
  participant ST as checkpoint 저장소
  participant K as Kafka
  JM->>SRC: checkpoint 42 시작 (barrier 주입)
  SRC->>ST: partition별 현재 offset 저장
  SRC->>SNK: 데이터 뒤에 barrier 흘려보냄
  SNK->>K: buffer flush (EXACTLY_ONCE면 transaction pre-commit)
  SNK->>ST: transaction 정보 저장
  ST-->>JM: 모든 operator 저장 완료
  JM->>JM: checkpoint 42 확정
  JM->>SRC: 완료 알림
  SRC->>K: offset commit (모니터링용)
  JM->>SNK: 완료 알림
  SNK->>K: transaction commit (EXACTLY_ONCE)

눈여겨볼 점은 Kafka에 무언가를 확정하는 일, 즉 offset commit과 transaction commit이 전부 checkpoint가 확정된 뒤에 일어난다는 것이다. checkpoint가 실패하면 Kafka 쪽에도 아무것도 확정되지 않는다.

4.2 Walking Through a Failure

TaskManager 하나가 checkpoint와 checkpoint 사이에 죽었다고 하자.

10:00:00  checkpoint 41 완료      partition 0 offset = 1000
10:00:00~ offset 1000~1300 처리, 결과를 sink로 내보냄
10:00:45  TaskManager 죽음        (checkpoint 42는 아직 시작 전)
10:00:47  재시작 → checkpoint 41 로드 → offset 1000부터 다시 읽음

그러면 offset 1000~1300은 두 번 처리된다. Flink는 이 중복 실행 자체를 막지 않는다. 대신 상태도 41번 시점으로 함께 되돌리기 때문에, Flink 안쪽에서는 한 번만 처리한 것과 같은 결과가 나온다.

정말 고민해야 할 곳은 Flink 바깥이다. 10:00:00~10:00:45 사이에 sink가 이미 내보낸 결과는 어떻게 되는가. 이 질문은 4.7에서 이어서 다룬다.

4.3 Starting Offsets Apply Only Without a Checkpoint

setStartingOffsets는 checkpoint가 없을 때만 쓰인다. checkpoint나 savepoint에서 복원할 때는 이 설정을 보지 않는다.

checkpoint가 없는 상황은 생각보다 자주 생긴다.

1. job을 처음 배포할 때
2. savepoint 경로를 빼먹는 등 checkpoint 없이 job을 새로 제출할 때
3. HA 없이 JobManager가 죽었다 살아났을 때(5장)

그래서 시작 위치를 고를 때는 실수로 처음부터 다시 시작하면 어디서부터 읽게 될지까지 생각해 둬야 한다. committedOffsets(EARLIEST)를 쓰고 commit.offsets.on.checkpoint를 켜 두면, checkpoint를 잃더라도 마지막으로 commit된 위치 근처부터 이어서 읽는다. 4.1에서 모니터링용이라고 했던 Kafka commit이 여기서 안전장치 역할을 한다.

4.4 Other Source Properties

partition 하나가 split 하나가 되고, split은 subtask들에 나눠 배정된다. 그래서 병렬도를 partition 수보다 크게 잡으면 남는 subtask는 할 일 없이 쉰다. producer가 batch 단위로 압축한(compression.type) record는 source 안의 Kafka consumer가 알아서 풀어 주므로 따로 처리할 필요가 없다.

4.5 One Term, Two Settings

Flink 문서에서 exactly-once라는 말은 서로 다른 두 설정에 나온다. 둘을 같은 것으로 보면 헷갈린다.

end-to-end exactly-once라고 말하려면 세 가지가 모두 맞아야 한다.

flowchart LR
  s["source<br/>다시 읽을 수 있어야 함"] --> f["Flink 상태<br/>checkpoint mode EXACTLY_ONCE"]
  f --> k["sink<br/>transaction 또는<br/>두 번 써도 결과가 같음"]
  k --> r["end-to-end exactly-once"]

이 중 하나라도 빠지면 전체 결과는 at-least-once가 된다.

4.6 Checkpoint Mode: Alignment or Not

3장의 alignment 예를 at-least-once mode와 나란히 놓으면 차이가 보인다.

input A:  a1 a2 [42] a3 a4
input B:  b1 b2 b3 b4 ........ [42]   ← 늦게 옴

EXACTLY_ONCE
  A의 barrier 42가 오면 a3, a4는 잠시 대기
  B의 barrier 42까지 오면 snapshot 42를 찍고 a3, a4 처리 재개
  → snapshot 42에는 barrier 앞의 record만 반영

AT_LEAST_ONCE
  기다리지 않고 a3, a4도 계속 처리
  B의 barrier 42가 오면 그때 snapshot 42를 찍음
  → snapshot 42에 a3, a4(원래 43번 몫)가 이미 반영됨
  → 42에서 복구하면 a3, a4를 다시 읽어 한 번 더 반영 = 중복

공식 문서가 정리한 요점은 이렇다.

• alignment 때문에 늘어나는 지연은 보통 몇 ms 수준이다. 모든 record를 꾸준히 몇 ms 안에 처리해야 하는 job만 AT_LEAST_ONCE를 고려하고, 대부분은 EXACTLY_ONCE가 낫다.
• alignment는 입력이 여러 개인 operator에서만 일어난다. 그래서 map, flatMap, filter만 이어진 job은 AT_LEAST_ONCE mode여도 실제로는 exactly-once로 동작한다.
• unaligned checkpoint는 EXACTLY_ONCE mode에서만 쓸 수 있다.

4.7 Sink Delivery Guarantee: What Downstream Sees

4.2의 장애 시나리오, 즉 offset 1000~1300을 다시 처리한 상황을 sink 설정별로 이어서 보자.

flowchart LR
  a["checkpoint 41 확정"] --> b["transaction T42 열림<br/>1000~1300 결과 기록"]
  b --> c["장애"]
  c --> d["T42 abort"]
  d --> e["41에서 재시작<br/>1000~1300 다시 처리"]
  e --> f["transaction T42'에 기록"]
  f --> g["checkpoint 42 확정 → T42' commit"]

EXACTLY_ONCE sink는 2단계 커밋(two-phase commit)으로 동작한다. checkpoint 때 transaction을 flush해 pre-commit하고(1단계), checkpoint가 확정됐다는 알림을 받으면 commit한다(2단계). AT_LEAST_ONCE와 EXACTLY_ONCE는 checkpoint가 켜져 있어야 동작한다. checkpoint가 꺼져 있으면 기다리거나 commit할 시점 자체가 없다.

4.8 The Cost of EXACTLY_ONCE

transaction에는 비용이 따른다.

• 결과가 늦게 보인다. commit은 checkpoint가 확정될 때 일어난다. interval이 1분이면 결과도 최대 1분 늦게 보인다.
• 읽는 쪽도 맞춰야 한다. Kafka consumer의 기본 isolation.level은 read_uncommitted라서 abort된 record까지 읽는다. 다운스트림 consumer를 read_committed로 두어야 중복이 보이지 않는다.
• transaction이 만료되면 데이터가 사라진다. Kafka는 transaction.timeout.ms가 지나도록 commit되지 않은 transaction을 버린다. 공식 문서는 이 값을 가장 긴 checkpoint 시간과 가장 긴 재시작 시간을 더한 것보다 크게 잡으라고 권한다. 다만 broker의 transaction.max.timeout.ms(Kafka 기본 15분)보다 크게 잡을 수는 없다.
• transactionalIdPrefix를 꼭 정해야 한다. 같은 Kafka cluster에서 도는 job끼리 이 값이 겹치면 서로의 transaction을 방해한다. job마다 고유한 값을 준다.

4.9 Choosing a Combination

transaction이 있어야만 exactly-once 효과를 얻는 것은 아니다. 같은 record를 두 번 써도 결과가 같게 만들 수 있다면 at-least-once로도 충분하다. id로 중복을 제거하거나 key로 upsert하는 경우가 그렇다. 이런 sink를 idempotent하다고 한다. 결국 마지막 구간의 정확성은 transaction이나 멱등성 중 하나로 확보한다.