체크포인트·세이브포인트와 Exactly-Once

장애가 나도 “같은 결과”로 돌아오게 만들기

스트리밍 잡은 보통 오래 실행된다. 몇 시간짜리 배치처럼 실패하면 처음부터 다시 돌리는 방식으로는 운영할 수 없다. TaskManager가 죽거나, 네트워크가 흔들리거나, sink 쓰기 중에 프로세스가 재시작되어도 파이프라인은 어느 지점까지 처리했는지와 그 시점의 상태가 무엇이었는지를 같이 복구해야 한다.

Apache Flink의 답은 체크포인트(checkpoint)다. 체크포인트는 잡 전체의 상태와 소스 위치를 주기적으로 저장한 일관된 스냅샷이다. 실패가 발생하면 Flink는 마지막으로 성공한 체크포인트에서 상태를 복원하고, Kafka 같은 replay 가능한 소스의 offset을 그 지점으로 되돌린 뒤 다시 읽는다.

중요한 점은 “이벤트가 물리적으로 한 번만 처리된다”는 뜻이 아니라는 것이다. 실패 뒤에는 일부 이벤트가 다시 읽힐 수 있다. Flink가 보장하려는 것은 각 이벤트가 Flink가 관리하는 상태와 최종 결과에 한 번 반영된 것처럼 보이게 하는 것이다.

체크포인트가 담는 것

체크포인트는 단순히 메모리 덤프가 아니다. 다음 두 가지가 서로 맞아야 한다.

• 각 source의 읽기 위치: Kafka partition offset, 파일 offset 등
• 각 stateful operator의 상태: window buffer, keyed state, timer, join state 등

예를 들어 user_id별 5분 매출 합계를 계산하는 잡이 있다고 하자. 체크포인트 N은 “Kafka offset 10,000까지 읽었고, 그때 각 user의 누적 합계는 이 값이다”라는 약속이다. 상태만 저장하고 offset을 저장하지 않으면 중복 처리 가능성이 커지고, offset만 저장하고 상태를 저장하지 않으면 집계 결과가 사라진다.

Barrier와 일관된 스냅샷

Flink는 분산 잡을 멈추지 않고 스냅샷을 만든다. 핵심 장치는 checkpoint barrier다. Checkpoint Coordinator가 source에 barrier를 넣으면 barrier는 일반 레코드와 같은 스트림을 따라 downstream operator로 흐른다.

Barrier는 “여기까지가 체크포인트 N에 포함될 데이터”라는 경계선이다. 단일 입력 operator는 barrier를 받았을 때 현재 상태를 snapshot하고 barrier를 다음 operator로 넘긴다. 여러 입력을 가진 operator는 더 까다롭다. 한 입력에서는 barrier가 도착했는데 다른 입력에서는 아직 도착하지 않았다면, aligned checkpoint에서는 먼저 도착한 입력 채널을 잠시 막고 나머지 barrier가 올 때까지 기다린다. 이렇게 해야 한쪽 입력의 checkpoint N 이후 데이터가 다른 쪽 입력의 checkpoint N 이전 데이터와 섞이지 않는다.

이 과정을 barrier alignment라고 한다. Backpressure가 심하거나 특정 파티션이 느리면 alignment 시간이 길어진다. 그래서 checkpoint 지연은 단순히 스토리지 문제만이 아니라, 데이터플로우 전체의 병목 신호일 수 있다.

체크포인트 설정에서 먼저 보는 값

운영에서 체크포인트는 “켜기만 하면 끝”이 아니다. 너무 자주 찍으면 상태 저장 I/O가 잡의 본업을 방해하고, 너무 드물게 찍으면 장애 후 되감아야 하는 데이터가 많아진다.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.enableCheckpointing(60_000); // 60초마다 체크포인트 시도

env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointTimeout(10 * 60 * 1000);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

기본적으로 Flink checkpointing은 비활성화되어 있으므로 job에서 명시적으로 켜야 한다. 운영 환경에서는 checkpoint storage를 S3, HDFS, GCS 같은 durable storage로 두는 것이 안전하다. JobManager heap 기반 저장은 로컬 실험이나 작은 상태에는 편하지만, HA 운영과 큰 state에는 맞지 않는다.

State backend와 checkpoint storage를 구분하기

Flink를 처음 운영할 때 자주 헷갈리는 지점이 state backend와 checkpoint storage다.

• State backend: 실행 중인 working state를 TaskManager 쪽에서 어떻게 들고 있을지 결정한다.
• Checkpoint storage: snapshot을 장애 복구용으로 어디에 영구 저장할지 결정한다.

HashMapStateBackend는 JVM heap에 state를 두므로 접근이 빠르지만, heap과 GC 영향을 받는다. EmbeddedRocksDBStateBackend는 local disk의 RocksDB에 state를 두므로 큰 state에 적합하지만 직렬화와 disk access 비용이 있다. RocksDB를 쓰는 큰 state job에서는 incremental checkpoint가 checkpoint 시간을 크게 줄일 수 있다. 다만 incremental checkpoint는 여러 checkpoint가 파일을 공유할 수 있으므로 저장소 정리와 retention 정책을 더 조심해야 한다.

Checkpoint와 Savepoint의 차이

둘 다 Flink job의 일관된 snapshot이지만 목적이 다르다.

체크포인트는 자동차의 에어백에 가깝다. 평소에는 보이지 않지만 사고가 나면 즉시 필요하다. 세이브포인트는 정비소에 들어가기 전 찍는 안전한 저장점에 가깝다. 코드를 바꾸거나 병렬도를 조정하거나 Flink 버전을 올릴 때 “여기서 멈추고, 새 job을 이 상태에서 시작하자”는 운영 절차에 쓴다.

Savepoint를 안전하게 쓰는 핵심: stable UID

Savepoint는 operator state를 Operator ID -> State 형태로 매핑한다. 그래서 stateful operator의 ID가 바뀌면 Flink가 예전 state를 새 operator에 연결하지 못할 수 있다.

events
    .keyBy(Event::userId)
    .process(new FraudDetector())
    .uid("fraud-detector-v1")
    .name("fraud-detector");

Flink는 UID를 자동 생성할 수 있지만, 자동 UID는 job graph 구조에 민감하다. 중간 operator를 하나 추가하거나 순서를 바꾸는 것만으로도 ID가 달라질 수 있다. 장기 운영할 job이라면 stateful operator에는 명시적인 uid(...)를 붙이는 편이 안전하다.

Savepoint restore에서는 다음 질문을 먼저 확인한다.

• 새 코드의 stateful operator UID가 기존 savepoint와 호환되는가?
• 삭제된 operator의 state를 어떻게 처리할 것인가?
• state serializer schema 변경이 호환되는가?
• restore 후 첫 checkpoint가 성공하기 전까지 원본 snapshot을 삭제하지 않았는가?

특히 restore claim mode에 따라 snapshot 파일의 소유권이 달라질 수 있다. 기본 NO_CLAIM 모드는 Flink가 복원에 사용한 snapshot을 소유하지 않고 사용자 관리로 남겨 둔다. 따라서 첫 번째 새 checkpoint가 성공하기 전까지는 원본 savepoint나 externalized checkpoint를 삭제하지 않는 것이 안전하다.

Exactly-Once는 어디까지 보장되는가

Flink의 exactly-once는 두 층으로 나누어 봐야 한다.

첫째, Flink 내부 상태의 exactly-once다. checkpoint와 barrier alignment 덕분에 operator state는 실패 전후에도 각 이벤트가 한 번 반영된 것처럼 복구된다.

둘째, end-to-end exactly-once다. 외부 시스템까지 포함해 최종 출력에서도 중복과 손실이 없어야 한다. 이를 위해서는 세 조건이 필요하다.

1. Source가 replay 가능해야 한다. Kafka처럼 checkpoint된 offset부터 다시 읽을 수 있어야 한다.
2. Flink job이 checkpoint 기반으로 state를 복구해야 한다.
3. Sink가 transactional write 또는 idempotent write를 제공해야 한다.

세 번째가 빠지면 Flink 내부 상태는 정확해도 외부 테이블이나 메시지 토픽에는 중복이 남을 수 있다. 예를 들어 sink가 HTTP API에 단순 POST를 보낸 뒤 실패하면, 복구 후 같은 이벤트가 다시 POST될 수 있다. 이 API가 멱등 키를 지원하지 않으면 end-to-end exactly-once가 아니다.

Two-phase commit과 sink의 책임

Transactional sink는 checkpoint와 commit 시점을 맞춘다. Flink의 checkpoint가 완료되기 전에는 외부 시스템에 “확정된 결과”로 노출하지 않고, checkpoint가 성공하면 해당 구간의 transaction을 commit한다. 실패하면 pre-commit된 transaction을 abort하거나, 복구 과정에서 완료되지 않은 transaction을 정리한다.

이 구조에서 가장 위험한 착각은 “Flink가 exactly-once니까 sink도 자동으로 exactly-once”라고 생각하는 것이다. 실제 운영에서는 sink connector 문서를 확인해야 한다. Kafka, 파일 sink, 일부 JDBC/upsert sink처럼 checkpoint와 맞물리는 의미가 분명한 경우도 있지만, 외부 API 호출이나 임의 DB insert는 별도 멱등 키, upsert key, transaction boundary를 설계해야 한다.

장애 복구 흐름

실패가 발생하면 Flink는 대략 다음 순서로 복구한다.

1. 실패한 job attempt를 중단한다.
2. 마지막으로 성공한 checkpoint metadata를 찾는다.
3. 각 operator의 state를 checkpoint storage에서 복원한다.
4. source 위치를 checkpoint에 기록된 offset으로 되돌린다.
5. 처리 중이던 미완료 sink transaction을 abort하거나 복구한다.
6. source에서 다시 읽으며 처리한다.

이때 checkpoint 이후 처리됐지만 아직 다음 checkpoint에 포함되지 않은 레코드는 다시 처리될 수 있다. 내부 state는 checkpoint 시점으로 돌아갔으므로 다시 처리되어도 결과는 한 번 반영된 것처럼 맞춰진다. 하지만 외부 sink가 checkpoint 이전에 이미 부작용을 노출했다면, 그 부작용은 Flink만으로 되돌릴 수 없다.

운영에서 보는 체크포인트 증상

체크포인트 문제는 보통 다음 모양으로 나타난다.

Unaligned checkpoint는 backpressure가 있는 상황에서 barrier가 더 빨리 지나가게 도와줄 수 있다. 그러나 근본 원인인 처리 병목을 없애는 기능은 아니다. end-to-end latency가 계속 높다면 unaligned checkpoint만 켜고 끝낼 문제가 아니라 병렬도, source rate, sink throughput, state access 비용을 함께 봐야 한다.

실전 배포 절차 예시

상태가 있는 Flink job을 새 버전으로 배포할 때는 다음 순서가 보수적이다.

1. 모든 stateful operator에 stable UID가 있는지 확인한다.
2. 현재 job의 checkpoint가 정상적으로 성공하고 있는지 확인한다.
3. stop-with-savepoint 또는 운영 도구로 savepoint를 만든다.
4. 새 job 버전이 같은 UID와 호환 가능한 state schema를 유지하는지 확인한다.
5. 새 job을 savepoint에서 restore한다.
6. 새 job의 첫 checkpoint가 성공할 때까지 원본 savepoint를 보존한다.
7. sink 출력에서 중복 또는 누락이 없는지 운영 지표로 확인한다.
8. 보존 정책에 따라 오래된 checkpoint/savepoint를 정리한다.

이 절차는 느려 보이지만, stateful streaming에서는 빠른 배포보다 되돌릴 수 있는 배포가 더 중요하다. 특히 window, join, deduplication, fraud detection처럼 state가 곧 비즈니스 의미인 job은 savepoint 호환성을 배포 체크리스트의 일부로 다뤄야 한다.

정리

체크포인트는 Flink의 장애 복구 장치이고, 세이브포인트는 운영자가 의도적으로 잡을 옮기거나 바꾸기 위한 저장점이다. 둘 다 snapshot이지만 수명과 목적이 다르다.

Exactly-once는 Flink 내부 상태만 보면 checkpoint와 barrier alignment로 설명할 수 있다. 그러나 end-to-end exactly-once는 source, Flink state, sink가 모두 참여하는 계약이다. replay 가능한 source, durable checkpoint storage, transactional 또는 idempotent sink 중 하나라도 빠지면 보장은 약해진다.

운영 관점의 핵심은 세 가지다.

• checkpoint interval, timeout, min pause를 처리량과 복구 목표에 맞게 조정한다.
• state backend와 checkpoint storage를 구분하고, 큰 state에서는 RocksDB와 incremental checkpoint를 검토한다.
• savepoint restore를 위해 stateful operator에 stable UID를 부여하고, sink의 exactly-once 의미를 문서로 확인한다.

References

• Apache Flink Docs: Checkpointing
• Apache Flink Docs: Fault Tolerance
• Apache Flink Docs: Checkpoints
• Apache Flink Docs: Savepoints
• Apache Flink Docs: Tuning Checkpoints and Large State
• Apache Flink Blog: An Overview of End-to-End Exactly-Once Processing in Apache Flink