5. Running Flink in Production

5.1 Where Config Lives

Flink 설정은 세 군데에 넣을 수 있다. 어디에 넣었는지에 따라 우선순위가 달라서, 이것부터 알아 두면 분명히 설정했는데 값이 왜 안 바뀌는지 헤매는 일이 줄어든다.

flowchart LR
  a["코드<br/>env.enableCheckpointing(...)<br/>env.configure(config)"] -->|"가장 우선"| r["실제 적용 값"]
  b["config.yaml<br/>Kubernetes Operator면 flinkConfiguration"] --> r
  c["Flink 기본값"] --> r

같은 key를 여러 곳에서 정했다면 코드, config.yaml, 기본값 순서로 적용된다. 코드에 적은 값이 가장 우선한다. 아래 두 블록은 같은 설정이다.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(1000);
env.getCheckpointConfig().setCheckpointingConsistencyMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
env.getCheckpointConfig().setCheckpointTimeout(60000);
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);

Configuration config = new Configuration();
config.set(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay");
config.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 3);
config.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY, Duration.ofSeconds(10));
env.configure(config);

execution.checkpointing.interval: 1s
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.min-pause: 500 ms
execution.checkpointing.timeout: 60 s
execution.checkpointing.tolerable-failed-checkpoints: 2
restart-strategy.type: fixed-delay
restart-strategy.fixed-delay.attempts: 3
restart-strategy.fixed-delay.delay: 10 s

5.2 When a Task Fails: Restart Strategies

TaskManager가 죽거나 operator에서 예외가 나면 Flink는 두 가지를 정해야 한다. 어디까지 다시 띄울지, 그리고 얼마나 기다렸다가 몇 번까지 다시 시도할지다.

flowchart TD
  f["task 실패"] --> s{"failover-strategy<br/>어디까지 다시 띄우나"}
  s -->|"region (기본)"| r1["영향받는 task만"]
  s -->|full| r2["job 전체"]
  r1 --> t{"restart-strategy.type<br/>얼마나 기다렸다 몇 번까지"}
  r2 --> t
  t -->|none| x["job 실패"]
  t -->|fixed-delay| a["같은 간격으로 N번<br/>넘으면 job 실패"]
  t -->|failure-rate| b["구간당 N번까지<br/>넘으면 job 실패"]
  t -->|exponential-delay| e["간격을 늘려 가며 무제한<br/>오래 정상이면 간격 초기화"]
  a --> c["마지막 checkpoint에서 재개"]
  b --> c
  e --> c

checkpoint를 켜 두고 재시작 전략을 따로 정하지 않으면 exponential-delay가 기본으로 적용된다.

exponential-delay를 기본값 그대로 두면 대기 시간이 이렇게 늘어난다.

1s → 1.5 → 2.3 → 3.4 → 5.1 → 7.6 → 11 → 17 → 26 → 38 → 57 → 60 → 60 → ...
                                                            ↑ max-backoff에서 멈춤
1시간(reset-backoff-threshold) 동안 문제없으면 다시 1초부터

jitter는 대기 시간에 약간의 무작위 값을 더한다. Kafka broker 재시작처럼 같은 원인으로 여러 job이 한꺼번에 죽었을 때, 모두 같은 순간에 다시 접속하지 않게 해 준다.

5.3 When the JobManager Fails: HA

5.2의 재시작 전략은 JobManager가 살아 있다는 전제로 동작한다. task를 다시 배치하고 어느 checkpoint에서 재개할지 고르는 일을 JobManager가 하기 때문이다. 그런데 JobManager는 기본적으로 하나뿐이다. 공식 문서는 이를 단일 장애 지점(single point of failure)이라고 부른다.

HA가 없을 때 어떤 일이 생기는지 4장의 내용과 이어서 보면 위험이 분명해진다.

HA 없음
  JobManager 죽음 → Kubernetes가 pod를 다시 띄움
  → 새 JobManager는 이전 job이 어느 checkpoint까지 갔는지 모름
  → job을 처음 제출하는 것처럼 시작
  → 상태는 비어 있고, KafkaSource는 setStartingOffsets로 시작
     earliest()면 topic 보관 기간 전체를 다시 읽음
     latest()면 죽어 있던 동안의 데이터를 건너뜀

HA 있음
  JobManager 죽음 → 새 JobManager가 HA 저장소에서 job과 최신 checkpoint 위치를 찾음
  → 그 checkpoint에서 재개 (TaskManager 장애 때와 같은 결과)

공식 문서는 HA 서비스의 역할을 셋으로 나눈다. leader election은 JobManager 후보 여러 개 중 하나만 leader가 되게 한다. service discovery는 TaskManager와 client가 지금 leader가 누구인지 찾게 한다. state persistence는 다음 leader가 job을 이어받는 데 필요한 것, 즉 JobGraph, 사용자 jar, 완료된 checkpoint 목록을 남겨 둔다.

구현은 ZooKeeper와 Kubernetes 두 가지다. ZooKeeper는 어디서나 쓸 수 있지만 quorum을 따로 운영해야 한다. Kubernetes HA는 Kubernetes에서만 쓸 수 있지만, ConfigMap으로 동작하므로 따로 띄울 것이 없다.

Where HA Data Is Stored

HA 정보는 두 곳에 나눠 저장된다. 큰 데이터는 파일 시스템에 두고, 그 위치를 가리키는 포인터만 Kubernetes ConfigMap(또는 ZooKeeper)에 둔다.

flowchart LR
  subgraph k8s["ConfigMap 또는 ZooKeeper"]
    l["현재 leader 주소"]
    p["포인터<br/>JobGraph 위치,<br/>최신 checkpoint 메타데이터 위치"]
  end
  subgraph sd["high-availability.storageDir"]
    jg["JobGraph, 사용자 jar"]
    cm["완료된 checkpoint 메타데이터"]
  end
  subgraph cd["execution.checkpointing.dir"]
    st["실제 state 파일"]
  end
  p --> jg
  p --> cm
  cm --> st

그래서 HA를 켤 때는 storageDir를 꼭 정해야 한다. ConfigMap에는 포인터만 있고 실제 내용은 storageDir에 있다.

Leader Failover on Kubernetes

sequenceDiagram
  participant JM1 as JobManager 1 (leader)
  participant CM as ConfigMap (lease)
  participant JM2 as JobManager 2
  participant SD as storageDir
  participant TM as TaskManager
  JM1->>CM: 주기적으로 lease 갱신
  Note over JM1: 장애
  JM2->>CM: retry-period마다 lease 확인
  Note over CM: lease-duration 동안 갱신 없음 → 만료
  JM2->>CM: lease 획득 (새 leader)
  JM2->>CM: 포인터 읽기
  JM2->>SD: JobGraph, 최신 checkpoint 메타데이터 로드
  TM->>CM: 새 leader 주소 확인 후 재연결
  JM2->>TM: 최신 checkpoint에서 task 재배치

기본값으로는 leader가 죽고 대략 lease-duration(15초)이 지나야 다른 후보가 leader가 된다. standby JobManager를 꼭 띄울 필요는 없다. kubernetes.jobmanager.replicas가 기본값 1이어도 Kubernetes가 pod를 다시 띄우고, 새 pod가 HA 정보를 읽어 이어받는다. replicas를 2 이상으로 두면 새 pod가 뜨기를 기다리지 않아도 되므로 복구가 빨라진다.

lease-duration을 줄이면 장애를 더 빨리 알아챈다. 대신 Kubernetes API가 잠깐 느려졌을 뿐인데도 leader가 바뀌는 일이 잦아진다.

When HA Data Is Cleaned Up

HA 데이터는 job이 끝날 때(성공, 취소, 최종 실패) 함께 지워진다. 반대로 job이 끝나지 않은 채 cluster만 내려가면 그대로 남는다.

Kubernetes 문서는 이 동작을 활용하는 방법도 소개한다. kubectl delete deployment <cluster-id>로 지우면 JobManager, TaskManager, 설정 ConfigMap은 지워지지만 HA ConfigMap은 남는다. 그래서 같은 cluster-id로 다시 띄우면 이전 job이 최신 checkpoint에서 자동으로 이어진다. 반대로 보면 이 동작은 함정이 되기도 한다. 깨끗하게 새로 시작하려고 cluster를 지웠다 띄웠는데 예전 job이 살아날 수 있다. 이럴 때는 남은 HA ConfigMap을 직접 지운다.

What Protects Against What

5.4 Deploying: Savepoints and uid

savepoint는 사람이 찍는 checkpoint다. 원리는 같지만 사용자가 직접 트리거하고 자동으로 지워지지 않는다. 코드를 고치거나, Flink 버전을 올리거나, 병렬도를 바꾸거나, 다른 cluster로 옮길 때처럼 멈추고, 고치고, 이어서 돌리는 작업에 쓴다.

코드를 고쳐 새 버전을 올릴 때는 먼저 savepoint를 찍고 job을 내린 뒤, 새 버전을 그 savepoint에서 시작한다.

flowchart LR
  a["job v1 실행 중"] -->|"savepoint 찍고 중지"| sp[("savepoint")]
  sp -->|"execution.state-recovery.path"| b["job v2 시작"]
  b --> m{"operator uid로<br/>상태 짝 맞추기"}
  m -->|"uid 같음"| ok["상태 이어받음"]
  m -->|"uid 바뀜, operator 삭제"| ng["복원 실패<br/>ignore-unclaimed-state면 건너뜀"]

여기서 자주 막히는 곳이 uid다. savepoint 안의 상태는 operator uid를 기준으로 새 job의 operator와 짝을 맞춘다. uid를 직접 정하지 않으면 Flink가 graph 구조를 보고 자동으로 만드는데, 이 값은 코드를 조금만 고쳐도 바뀔 수 있다. 그러면 새 job이 예전 상태를 찾지 못한다. 공식 문서도 운영 배포 전에 uid를 직접 정해 두라고 강하게 권한다.

stream.map(new MyMapper()).uid("my-mapper").name("My mapper");   // uid는 상태 짝, name은 UI 표시용

max-parallelism을 넘지 않는다면 savepoint에서 병렬도를 바꿔 띄울 수 있다. 처리량이 모자라 병렬도를 늘릴 때도 이 방법을 쓰며, 이때 상태는 key group 단위로 재분배된다.

5.5 Latency vs. Throughput

1장에서 본 buffer와 timeout 사이의 절충을 조절하는 key들이다. buffer를 오래 모아 보낼수록 처리량이 오르고, 자주 보낼수록 지연이 줄어든다.

5.6 Monitoring

JobManager와 TaskManager가 metric을 노출하면 Prometheus가 수집하고, Grafana와 알림이 그 위에 붙는 구성이 흔하다.

flowchart LR
  jm["JobManager :9249"] --> p[("Prometheus")]
  tm["TaskManager :9249"] --> p
  p --> g["Grafana, 알림"]

먼저 볼 metric은 이 정도다.

checkpoint가 느려졌다면 먼저 3장에서 본 두 원인 중 어느 쪽인지 가린다. lastCheckpointSize가 크면 상태가 큰 것이고, backPressuredTimeMsPerSecond가 높은 operator가 있으면 barrier가 늦게 도착하는 것이다. 상태가 크면 증분 checkpoint로 풀고, barrier가 늦으면 unaligned checkpoint나 buffer debloating을 쓰거나 병목 operator의 병렬도를 늘려서 푼다.