이 문서를 읽고 나면 다음 질문에 답할 수 있다.
- Chandy-Lamport 알고리즘이 풀려는 문제는 무엇인가? "전체를 멈추지 않고" 스냅샷을 찍는다는 게 왜 어려운가?
- "일관된 글로벌 상태(consistent global state)"는 정확히 무엇으로 정의되는가?
- 원본 Chandy-Lamport와 Flink의 비동기 barrier snapshot(ABS)는 같은 알고리즘인가, 다른 알고리즘인가?
- 왜 Flink는 "marker(barrier)"를 스트림 안에 흘려보내는가? 외부에서 신호를 보내면 안 되는가?
- 이 알고리즘이 정확히 어떤 "일관성"을 보장하고, 어떤 일관성은 보장하지 않는가?
여러 노드가 서로 메시지를 주고받는 분산 시스템이 있다고 하자.
msg A msg B
TM1 ─────────▶ TM2 ─────────▶ TM3
이 시스템의 "지금 이 순간 상태"는 무엇인가?
단순한 발상:
"모두에게 '상태 보고' 메시지를 보내고, 받은 응답을 모은다"
문제:
TM1에서 보낸 msg A가 아직 TM2에 안 도착한 시점에 TM2의 상태를 찍었다고 하자.
조금 뒤 TM2가 msg A를 받아 자기 상태를 업데이트하고 TM3에 msg B를 보낸다.
TM3는 msg B를 처리한 후의 상태를 보고한다.
→ 보고된 스냅샷: TM1 보냄, TM2 안 받음, TM3 그 결과 반영됨
→ "유령 메시지"가 존재 — TM3가 처리한 msg B는 어디서 왔는가?
→ 이걸로 복원하면 시스템이 인과율을 어긴 상태가 된다
Chandy-Lamport(1985)의 통찰:
- 시계를 동기화할 필요가 없다 (시계는 신뢰할 수 없으니)
- 시스템을 멈출 필요도 없다 (가능하지 않으니)
- 대신 "마커(marker)" 메시지를 정상 데이터 흐름 안에 끼워 넣어
각 노드가 "마커를 받은 순간"의 상태를 찍게 한다
- 채널의 메시지도 일부 기록하여 in-flight를 보존한다
Flink 잡:
Source → Map → KeyBy → Window → Sink
│ │ │ │ │
│ │ │ │ │
TaskMgr 여러 대에 분산, 각 sub-task가 자기 상태 가짐
sub-task 사이는 네트워크 채널로 데이터 흐름
질문:
"이 잡의 현재 일관된 상태"는 무엇이고,
처리를 멈추지 않고 어떻게 찍을 것인가?
답: Chandy-Lamport 영감의 변형 알고리즘
→ "Lightweight Asynchronous Snapshots for Distributed Dataflows"
(Carbone et al., 2015) — Flink 체크포인트의 토대 논문
// 잘못된 이해
// "Flink는 Chandy-Lamport를 그대로 구현했다"
// 실제:
// 영감과 핵심 아이디어(마커 기반)는 같지만 둘은 다른 알고리즘이다.
//
// Chandy-Lamport (원본):
// - 임의 토폴로지의 분산 시스템 대상 (사이클 허용)
// - 각 노드가 임의 시점에 스냅샷 시작 가능
// - "사후"에 in-flight 메시지를 채널에 기록해서 보존
//
// Flink ABS (Asynchronous Barrier Snapshotting):
// - DAG(방향성 비순환 그래프) 토폴로지에 특화
// - JobManager(중앙)가 barrier를 트리거
// - aligned 변종에서는 in-flight 메시지 보존이 필요 없다
// (각 채널 입력이 같은 barrier까지 정렬되면 자동으로 일관됨)
// - unaligned 변종에서는 in-flight 메시지를 명시적으로 기록 (Ch5-03)// 잘못된 이해
// "barrier는 '지금 스냅샷 찍어!' 신호일 뿐, 위치는 중요하지 않다"
// 실제:
// barrier의 정확한 위치(데이터 스트림 안에서 어느 이벤트들 사이에 있는가)가
// 스냅샷의 "정확히 어느 시점"을 정의한다.
//
// barrier 앞에 있는 이벤트들 = "이 체크포인트에 포함된 이벤트"
// barrier 뒤에 있는 이벤트들 = "다음 체크포인트에 포함될 이벤트"
//
// 이게 일관성의 핵심: 어떤 이벤트가 어디에 속하는지 명확히 갈리고,
// "걸쳐 있는" 이벤트가 없다.// 잘못된 이해
// "일관된 글로벌 상태란 모든 노드가 같은 wall-clock 시각의 상태를 갖는 것"
// 실제:
// wall-clock과는 관계 없다. 정의는 "원인-결과 관계"에 기반한다.
//
// 일관된 글로벌 상태 = "기록된 어떤 메시지 수신도, 같은 기록된 시점 이전에
// 그 메시지가 송신되었어야 한다"
//
// 즉 인과율(causality)을 만족하는 시점.
// 두 노드가 wall-clock 12:00:00.000과 12:00:00.500의 상태를 찍었더라도,
// 그 사이에 흐른 메시지가 모두 일관되게 잡혔으면 그건 valid한 글로벌 상태다.시스템 모델:
- 여러 프로세스 P1, P2, ..., Pn
- 프로세스 간 FIFO 채널들 (예: c12, c13, c23 ...)
- 채널은 신뢰성 있고 순서 보존, 지연은 임의
알고리즘 (initiator가 P1이라고 가정):
(1) P1: 자기 상태를 기록 (snapshot S1)
P1: 모든 outgoing 채널에 marker 전송
(2) 임의의 Pi가 채널 c_ji에서 marker를 처음 받았을 때:
→ 자기 상태를 기록 (snapshot Si)
→ 채널 c_ji의 기록 = 공집합 (marker 전 메시지는 다 처리됨)
→ 다른 incoming 채널들에 대해 "marker 도착까지 받은 메시지 기록 시작"
→ 자기 모든 outgoing 채널에 marker 전송
(3) 그 다음부터 Pi가 다른 incoming 채널 c_ki에서 marker를 받으면:
→ 그동안 c_ki에서 기록한 메시지들 = 그 채널의 state로 보존
(4) 모든 Pi가 모든 incoming 채널의 marker를 받으면 알고리즘 종료
글로벌 스냅샷 = {S1, S2, ..., Sn} ∪ {모든 채널 상태}
핵심 보장: 이 글로벌 스냅샷은 일관성을 만족한다. 즉 어떤 실제 실행이 있었더라도 시스템이 도달할 수 있었던 한 상태와 동등하다.
Flink의 DAG에 특화된 단순화:
[1] JobManager가 모든 source의 sub-task에 "checkpoint N 시작" 트리거
[2] 각 source가:
- 자기 상태(Kafka offset 등)를 비동기로 외부 저장소에 업로드 시작
- 자기 outgoing 스트림에 "barrier N"을 다음 레코드로 삽입
[3] 다운스트림 operator가 incoming 채널들에서 barrier N을 받는다
여러 채널이 있으면 (예: keyBy 후) — 모든 채널에서 barrier N이 다 도착할 때까지
먼저 도착한 채널의 후속 데이터는 버퍼링 (이게 "alignment", Ch5-03)
모든 채널에서 barrier N 도착 → 자기 상태 스냅샷 → barrier N을 outgoing 전송
[4] sink까지 barrier가 도달, 각 operator가 JobManager에 ACK
[5] 모든 ACK 수신 → 체크포인트 N 완료
원본 vs Flink ABS 비교:
┌────────────────────────┬─────────────────────┬─────────────────────────┐
│ 항목 │ Chandy-Lamport │ Flink ABS │
├────────────────────────┼─────────────────────┼─────────────────────────┤
│ 트리거 │ 임의 노드 │ JobManager(중앙) │
│ 토폴로지 │ 임의 (사이클 OK) │ DAG (사이클 X*) │
│ in-flight 메시지 기록 │ 항상 필요 │ aligned: 불필요 │
│ │ │ unaligned: 필요 (제한적)│
│ 채널 모델 │ FIFO + 신뢰성 │ FIFO + 신뢰성 │
│ 정렬 │ 없음 (각자 진행) │ aligned mode에서 있음 │
│ 외부 저장소 │ 메모리/디스크 │ S3/HDFS/MinIO 등 │
└────────────────────────┴─────────────────────┴─────────────────────────┘
* 사이클을 가진 잡(iterative dataflow)을 위한 별도 변형이 있다
정의 (정형):
실행 E = 메시지 송수신과 상태 변화의 시간 순서 시퀀스
글로벌 상태 G = {각 프로세스 상태, 각 채널 상태}
G가 일관됨 ⟺ 다음을 만족하는 실행 E'가 존재:
E에서 발생한 어떤 이벤트 e가 G에 기록되어 있으면,
E에서 e보다 인과적으로 앞선 모든 이벤트 e'도 G에 기록되어 있음
쉽게 말하면:
"송신은 기록되었는데 수신은 기록 안 됨" → 채널 상태에 그 메시지가 있어야 함
"수신은 기록되었는데 송신은 기록 안 됨" → 절대 일어나서는 안 됨 (불일치)
Chandy-Lamport는 두 번째 케이스가 절대 발생하지 않음을 증명한다. FIFO 채널 + marker 알고리즘의 결합이 그 보장의 핵심이다.
가정: TCP 같은 FIFO 채널
P1 → P2: 메시지 m1, m2 순서로 전송
P1 → P2: marker 전송 (m1, m2 이후)
→ P2는 m1, m2를 먼저 받고 그 다음 marker를 받는다
→ marker 전에 받은 메시지는 모두 "체크포인트 이전"에 처리됨이 보장
만약 채널이 FIFO가 아니라면:
P2가 marker를 m2보다 먼저 받을 수 있다
→ P2는 marker 시점에 m2를 아직 못 받았으므로 자기 상태에 반영 안 됨
→ 그러나 P1은 m2를 m1과 marker 사이에 보냈으니 자기 상태에는 반영 후
→ 글로벌 스냅샷에 "송신 기록, 수신 미기록" 발생, 그러나 채널에도 안 잡힘
→ 일관성 깨짐
Flink의 TCP 기반 네트워크 스택이 FIFO를 보장하기 때문에 이 알고리즘이 동작한다. 만약 UDP 같은 무순서 채널을 썼다면 동일 알고리즘으로는 정확성을 보장할 수 없다.
// Source operator의 시각
class SourceOperator {
void onCheckpointTrigger(long checkpointId) {
// 1) 자기 상태 (offset) 스냅샷 시작
StateSnapshot snap = stateBackend.snapshot(checkpointId);
// 2) outgoing 스트림에 barrier 삽입
for (Channel out : outgoingChannels) {
out.emit(new CheckpointBarrier(checkpointId));
}
// 3) 비동기로 외부 스토리지 업로드
asyncIO.upload(snap, () -> {
// 4) 완료되면 JobManager에 ACK
jobManager.acknowledgeCheckpoint(checkpointId, this, snap.getHandle());
});
}
}
// 일반 operator의 시각
class StreamOperator {
Map<Long, Set<Channel>> awaitingBarriers = new HashMap<>();
Map<Channel, Queue<Record>> blockedQueues = new HashMap<>();
void onRecord(Record r, Channel from) {
if (r instanceof CheckpointBarrier) {
long ckptId = ((CheckpointBarrier) r).id;
// 이 채널에서 barrier 도착 처리
awaitingBarriers.get(ckptId).remove(from);
if (awaitingBarriers.get(ckptId).isEmpty()) {
// 모든 채널에서 barrier 도착 → 자기 상태 스냅샷
StateSnapshot snap = stateBackend.snapshot(ckptId);
// 다운스트림으로 barrier 전송
for (Channel out : outgoingChannels) out.emit(r);
// 막아 두었던 큐 해제
unblockAllChannels();
// 비동기 업로드 + ACK
asyncIO.upload(snap, () -> jobManager.acknowledgeCheckpoint(ckptId, this, snap.getHandle()));
} else {
// 일부 채널에서만 barrier 도착 → 그 채널의 이후 데이터는 막아 두기
blockChannel(from);
}
return;
}
// 일반 레코드 처리
if (isBlocked(from)) blockedQueues.get(from).add(r);
else processRecord(r);
}
}순진한 발상은 "JobManager가 모든 task에 RPC로 '지금 멈춰서 상태 저장!' 명령을 내리는 것"이다. 왜 안 되는가?
[1] 동시 시점이 존재하지 않는다
JobManager가 모두에게 RPC를 보내도 도착 시점은 다 다르다
각자 "도착한 순간"의 상태를 찍으면 그게 일관되지 않다 (위의 인과율 문제 그대로)
[2] in-flight 데이터가 보존되지 않는다
RPC 도착 시점에 채널을 흐르고 있던 메시지들은 어디로 가는가?
송신자는 이미 보냈고, 수신자는 아직 처리 안 했고, 두 상태 모두 기록 안 됨
→ 복원 후 그 메시지는 영영 사라짐
[3] 데이터 흐름을 멈춰야 하는 경우 평소 처리 무너짐
해결: marker를 데이터 흐름 안에 끼워 넣으면,
"데이터 흐름의 자연스러운 순서"가 일관성을 보장하는 도구가 된다.
외부 동기화 없이 분산 일관성을 얻는 우아한 방법.
이게 Chandy-Lamport의 천재적 통찰이고, Flink가 그대로 계승한 부분이다.
Iterative 잡:
Source → Map → Loop ───┐
▲ │
│ │
└───────┘
(back edge)
이런 잡에서는 단순 ABS가 멈추지 않을 수 있다:
- back edge로 흐르는 메시지가 영원히 다음 체크포인트로 미뤄질 수 있음
- back edge에 in-flight 메시지가 모이면 alignment가 끝나지 않음
Flink의 해결:
back edge를 명시적으로 표시 (IterativeStream)
그 채널의 in-flight 메시지는 별도로 기록 (Chandy-Lamport 원본에 가까운 방식)
→ 사이클 있는 잡도 체크포인트 가능
대부분의 운영 잡은 DAG라서 이 복잡함은 없지만, ML 추론 피드백 루프 같은 잡에서는 고려해야 한다.
// BarrierFlowJob.java
public class BarrierFlowJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
env.enableCheckpointing(5_000);
env.getCheckpointConfig().setCheckpointStorage("file:///tmp/checkpoints");
DataStream<Long> source = env.fromSequence(1, Long.MAX_VALUE)
.map(x -> { Thread.sleep(5); return x; }).name("slow-map");
source
.keyBy(x -> x % 4)
.sum(0)
.map(x -> { Thread.sleep(10); return x; }).name("slow-sink")
.print();
env.execute("barrier-flow");
}
}# 잡 제출 후 Flink UI에서:
http://localhost:8081 → 잡 → Checkpoints → History 탭
각 체크포인트 클릭 시 "Subtasks" 탭:
task별로 barrier 도착 시각, 동기 단계 시작, 비동기 시작/끝 모두 보인다
→ barrier가 source에서 sink로 흐르는 시간이 어디서 얼마나 걸리는지 확인// 한 키를 의도적으로 hot하게 만들어 alignment 시간이 늘어나는 것을 본다
DataStream<Long> hotSkewed = env.fromSequence(1, Long.MAX_VALUE)
.map(x -> x % 100 == 0 ? Thread.sleep(50) : Thread.sleep(1));Flink UI → Checkpoints → 각 체크포인트의 "Alignment Duration"이 길어진다.
// 두 stateful operator에 카운터를 넣고
// "operator A의 카운터 + operator B의 카운터 = 처리한 총 이벤트 수"가
// 어느 체크포인트에서도 성립해야 한다
class CountingMap extends RichMapFunction<Long, Long> {
transient ValueState<Long> count;
// ...
@Override
public Long map(Long v) {
long c = (count.value() == null ? 0L : count.value()) + 1;
count.update(c);
return v;
}
}
DataStream<Long> stream = env.fromSequence(1, 1_000_000);
stream.map(new CountingMap()).name("A")
.keyBy(x -> x % 10)
.map(new CountingMap()).name("B")
.addSink(new DiscardingSink<>());
// 체크포인트에서 복원 후:
// A의 keyed state 합 == B의 keyed state 합 이어야 일관성 성립
// (실험적으로 KILL 후 재시작해서 확인)실험: 4-stage DAG (Source → Map → KeyBy+Window → Sink), 병렬성 8
Kafka 입력 200K rec/s, RocksDB, S3 저장, alignment mode
체크포인트 단계별 소요 시간:
Trigger 전파 (JM → 모든 source) : ~5 ms
Source 상태 스냅샷 시작 : ~10 ms (offset만 저장)
Source → Map barrier 흐름 : ~50 ms (네트워크 + 큐)
Map alignment + 스냅샷 : ~80 ms (작은 상태)
Map → Window barrier 흐름 : ~100 ms
Window alignment + 스냅샷 : ~600 ms (큰 상태 — RocksDB SST)
Window → Sink barrier 흐름 : ~150 ms
Sink alignment + commit prep : ~200 ms
비동기 업로드 (S3) : ~2000 ms (병렬 업로드)
모든 ACK 수신 : ~50 ms
─────────────────────────────────────────────
Total E2E : ~3.2 sec
주목:
- 동기 단계 (작업 잠시 멈춤) = alignment + sync snapshot = ~900 ms
- 비동기 단계 (백그라운드) = ~2000 ms
- 처리량 영향은 동기 단계에서만 발생
체크포인트 간격 5초로 설정 시 → 80% 시간은 정상 처리, 20%는 alignment 영향
중앙 트리거 (Flink ABS):
✅ 단순한 조정 (모든 source가 같은 ID를 같은 시점에 받음)
✅ 진행 상황을 JobManager가 일원 관리
❌ JobManager가 SPOF (HA로 완화)
❌ 큰 잡에서 트리거 RPC가 모든 source에 동시 전파되어 부하 spike
분산 트리거 (원본 Chandy-Lamport):
✅ initiator 자유, 부분 스냅샷 가능
❌ 여러 동시 스냅샷의 조정이 복잡
❌ 글로벌 ACK 모으는 메커니즘 별도 필요
aligned vs unaligned (Ch5-03 미리보기):
aligned: in-flight 데이터 보존 X, alignment 시간 비용
unaligned: in-flight 데이터 보존 O, alignment 거의 없음, 체크포인트 크기↑
FIFO 채널 의존성:
✅ 알고리즘이 단순해짐 (채널 별도 기록 없이 일관성 보장 가능)
❌ 채널 구현이 무조건 FIFO여야 함 (Flink는 TCP라 자연스러움)
이 알고리즘이 보장하지 않는 것:
- 외부 시스템(Kafka, DB)의 일관성: 그건 싱크 의미론의 책임 (Ch5-04, 05)
- 시계 동기화: 안 한다, 안 필요하다
- 실시간성: 스냅샷이 wall-clock 어느 시각에 해당하는지 정의되지 않음
Chandy-Lamport (1985)
목적: 분산 시스템의 일관된 글로벌 상태를 시스템을 멈추지 않고 찍기
방법: marker를 데이터 흐름에 끼워넣고, 각 노드가 marker 도착 시점에 자기 상태 기록
보장: 인과율을 만족하는 글로벌 상태 (송신 기록 없이 수신 기록되는 일은 없음)
요구: FIFO 채널
Flink ABS (Carbone et al., 2015)
Chandy-Lamport의 DAG-특화 변형
중앙(JobManager) 트리거 + barrier가 흐름 안에 흐름
aligned mode에서는 in-flight 메시지 보존 불필요 (alignment가 대신함)
unaligned mode에서는 명시적 in-flight 보존 (다음 문서)
"일관된 글로벌 상태"의 정의
wall-clock과 무관, 인과율 기반
"송신 기록 없이 수신만 기록"이 없으면 일관됨
Flink ABS 모두 이 의미의 일관성을 보장
barrier의 본질
단순 신호가 아니라 데이터 스트림 안의 marker
"barrier 앞에 있는 이벤트 = 이 체크포인트에 포함"이라는 경계
이 경계가 명확하게 갈리는 것이 일관성의 토대
다음 챕터 예고
03 → barrier가 여러 채널에서 올 때 어떻게 정렬하는가 (aligned/unaligned 차이)
04 → 이 일관된 상태가 외부 효과의 Exactly-Once로 어떻게 이어지는가
연결 학습: distributed-systems-theory-deep-dive의 Chandy-Lamport 분산 스냅샷 챕터와 함께 보면 이 알고리즘의 일관성 증명까지 따라갈 수 있다. 거기서 다루는 happens-before 관계, FIFO 가정의 역할이 Flink에 그대로 적용된다.
Q1. Chandy-Lamport는 채널이 FIFO여야 한다는 가정에 의존한다. 만약 Flink가 UDP 같은 무순서 채널을 사용해야 한다면, 동일 보장을 얻으려면 알고리즘에 어떤 변경이 필요한가?
Q2. Flink는 JobManager가 중앙에서 barrier 트리거를 한다. 만약 JobManager가 죽으면 진행 중인 체크포인트는 어떻게 되는가? 새 JobManager가 일어났을 때 잡의 정합성은 어떻게 보장되는가?
Q3. "Chandy-Lamport는 일관된 글로벌 상태를 보장한다"는 명제와 "Flink는 Exactly-Once를 보장한다"는 명제는 본질적으로 같은 말인가? 다르다면 무엇이 추가로 필요한가?
💡 해설
Q1. FIFO가 깨지면 marker가 정상 메시지보다 늦게 도착할 수 있다. 그 사이에 P2가 m2를 받지 못한 상태로 marker를 처리해 스냅샷을 찍으면 일관성이 깨진다. 해결책 두 가지가 있다. (1) 채널 수준 시퀀싱: 각 메시지에 시퀀스 번호를 붙이고, marker에도 "이 marker는 시퀀스 N까지를 포함한다"는 정보를 담아, 수신자는 N까지의 모든 메시지를 받은 후에야 스냅샷을 찍는다. (2) 확장된 채널 기록: marker 받는 시점을 기준으로 "이미 보냈는데 아직 안 받은 메시지"를 송신자가 직접 기록해 채널 상태로 보존한다. 어느 쪽이든 알고리즘이 복잡해지고 오버헤드가 늘어난다. 사실상 Flink가 TCP를 선택한 이유 중 하나가 이 단순성이다.
Q2. Flink의 JobManager는 high-availability(HA) 모드에서 ZooKeeper나 Kubernetes 리더 선출로 백업된다. JobManager가 죽으면 (1) 새 JM이 리더로 선출되고 (2) HA 저장소에서 마지막 완료된 체크포인트의 메타데이터를 읽고 (3) 그 체크포인트에서 잡을 재시작한다. 진행 중이던 (아직 ACK 모두 못 받은) 체크포인트는 자동으로 폐기된다. 정합성은 "완료된 체크포인트만 HA 저장소에 등록한다"는 원자성에서 나온다. 만약 N번 체크포인트가 ACK를 모두 받은 직후, HA 저장소에 등록되기 전에 JM이 죽으면? 이 경우엔 N은 완료된 것으로 안 보이고, 새 JM은 N-1에서 시작한다 — 약간의 재처리는 있지만 일관성은 유지된다. 결국 "원자적 완료 등록 + HA에서의 마지막 완료된 것 읽기"가 핵심.
Q3. 다른 명제다. Chandy-Lamport는 "내부 상태의 일관된 스냅샷"만 보장한다. 이건 필요조건이지 충분조건이 아니다. Exactly-Once에는 추가로 두 가지가 필요하다. (1) 재생 가능한 소스: 장애 후 마지막 체크포인트의 offset부터 데이터를 다시 읽을 수 있어야 한다. Kafka는 가능, 단발성 큐는 불가능. (2) 트랜잭션/멱등 싱크: 마지막 체크포인트 이후 외부에 이미 쓴 효과를 어떻게 할 것인가. 멱등하면 중복 쓰기 OK, 트랜잭션이면 rollback, 둘 다 아니면 Exactly-Once 못 함. Flink가 자랑하는 "end-to-end Exactly-Once"는 (Chandy-Lamport 스냅샷) + (재생 가능 소스) + (트랜잭션/멱등 싱크)의 결합이다. Ch5-04와 05에서 정확히 이 결합을 풀어낸다.