.
지난 글까지 Flink가 State를 Checkpoint/Savepoint로 외부에 스냅샷하고 Kafka offset도 그 State 안에 함께 넣어 관리한다는 것까지 정리했다. 이번엔 그 스냅샷이 어떻게 새 Job에 다시 붙는지를 다룬다.
State가 복원된다는 것은 스냅샷 안의 조각 하나하나가 새 Job의 operator 하나하나에 매칭된다는 뜻이다. 이 매칭 규칙을 모르면 복원은 성공했는데 데이터가 엉키는 일이 생긴다. 소스 토픽 리스트 중간에 토픽 하나를 끼워 넣었더니 뒤쪽 토픽 8개가 전부 한 칸 앞 토픽의 데이터를 받기 시작한 사건을 겪었고, 그 과정에서 확인한 것들을 정리한다.
Savepoint 메타데이터: operator ID를 키로 하는 맵
Savepoint 디렉터리의 _metadata 의 구조:
_metadata
├── magic / 포맷 버전 / checkpoint id
└── operator state × N
├── [operator ID: 16바이트] → 이 operator의 state 위치·크기
├── [operator ID: 16바이트] → …
└── …
Plain Text
복사
•
state에 operator ID 키로 저장
•
Job 이름, .name()으로 붙인 operator 이름, 토픽명, 그래프상의 위치는 매칭에 쓰이지 않음
복원 시 Flink가 하는 일:
1.
새 Job의 operator마다 ID를 계산
2.
_metadata의 ID 목록에서 같은 ID를 찾음
3.
있으면 그 state를 붙이고 없으면 빈 state로 시작
복원 과정에서 ID가 바뀌거나 없어지면 state는 주인을 잃고 ID가 우연히 맞으면 남의 state가 붙을 수 있다.
Operator ID 생성: uid 지정과 자동 해시
uid 있음
•
uid 문자열이 정해지면 그 operator의 ID도 고정. 배포를 몇 번 하든, 그래프에서 위치가 바뀌든 동일
•
로컬에서 mmh3.hash_bytes(uid, seed=0)로 계산한 값과 Flink UI의 vertex ID가 일치하는 것으로 직접 확인 가능
uid 없음 → ID는 자리 번호
•
StreamGraphHasherV2 가 그래프를 소스부터 순회하며 지금까지 해시한 노드 수와 입력 노드의 해시를 재료로 생성
•
코드가 한 글자도 안 바뀌어도 노드 순서가 바뀌면 ID가 바뀜
•
반대로 자리가 같으면 내용이 달라도 ID가 같음. 다른 Job이라도 그래프 모양이 같으면 operator ID가 완전히 동일
◦
예시:
clean_a 잡: source(topic-a) → map → sink
clean_b 잡: source(topic-b) → map → sink
Plain Text
복사
◦
읽는 토픽도 다르고 배포도 따로지만 노드 구성이 동일 → operator ID가 바이트 단위로 같음
◦
clean_a의 savepoint를 clean_b에 물려도 ID가 전부 맞으므로 Flink는 정상 복원으로 처리
◦
경고도 실패도 없이 clean_b의 source가 topic-b 대신 topic-a의 파티션·offset을 이어 읽음
트러블슈팅 사례: 키 하나 추가 후 뒤 토픽 8개가 밀린 일
상황
•
여러 원천 토픽을 내부 Kafka로 넘기는 멀티 소스 잡. Kafka source가 23개
•
코드는 토픽 키 리스트를 순회하며 키마다 source → process → side output 두 갈래 → sink 두 개 체인을 생성
KEYS = [..., "view", "click", "search", ..., "order", "point", "payment"]
for key in KEYS:
src = env.from_source(kafka_source(key), ..., f"{key}_source")
out = src.process(Passthrough(), ...).name(f"process_{key}")
out.get_side_output(ok_tag).sink_to(sink(key))
out.get_side_output(err_tag).sink_to(dlt_sink(key))
Python
복사
•
새 토픽 하나를 추가하면서 키를 리스트 중간(view 바로 뒤)에 삽입 → savepoint 모드로 배포
•
배포 성공, Job은 RUNNING
•
그런데 하류에서 order 토픽에 point 레코드가, point 토픽에 payment 레코드가 섞여 나옴
•
삽입한 키부터 뒤쪽 키 전부가 한 칸 앞 키의 데이터를 받고 있었음
원인: ID가 자리 번호
이 잡은 operator에 uid를 붙이지 않고 운영되고 있었다.
•
노드 하나가 삽입되면서 뒤 operator들의 자리가 밀림 → ID가 바뀜
•
자리 번호는 배포가 바뀌어도 같은 규칙으로 계산되므로, 밀려서 새로 받은 ID가 하필 옛 이웃 operator의 ID와 동일
자리 | 배포 전 | 배포 후 | 배포 후 operator가 받은 state |
p | view | view | view — 정상 |
p+1 | click | NEW | click - 새 토픽 source가 click 토픽을 읽음 |
p+2 | search | click | search |
… | … | … | 한 칸씩 밀림 |
p+7 | point | order | point |
p+8 | payment | point | payment |
p+9 | — | payment | 없음 → 버려짐 |
•
삽입한 키부터 뒤쪽 전부가 이 상태
•
맨 끝 payment는 자리가 하나 늘어 대응되는 옛 state가 없음
•
savepoint 모드든 last-state든 HA 복구든 붙는 기준은 operator ID 하나라 경로를 바꿔도 결과는 같음
배포가 조용히 성공한 이유
복원 시 ID 대조 결과는 세 가지:
경우 | 동작 |
양쪽에 있음 | state 복원 |
새 Job에만 있음 | 빈 state로 시작. 항상 허용 |
Savepoint에만 있음 | allowNonRestoredState=false면 복원 실패, true면 해당 state를 버리고 시작 |
이번 배포에서 세 경우가 다 나왔다.
•
밀린 키들 → 양쪽에 있음. ID가 전부 맞았으므로 정상 복원
•
새로 넣은 키 → 새 Job에만 있음. 빈 state로 시작
•
맨 끝 payment 자리의 옛 state → savepoint에만 있음. true라 조용히 폐기
세 경우 모두 Flink 입장에서는 정상 동작이라 경고가 나올 곳이 없었다.
자기 토픽까지 같이 읽은 이유
Kafka source가 무엇을 읽을지는 두 곳에서 정해진다.
•
state: reader가 복원한 split 목록. split에는 (topic, partition, offset)이 들어 있음
•
설정: 코드에 쓴 토픽명. enumerator가 주기적으로 이 토픽의 파티션을 스캔 (partition.discovery.interval.ms, Kafka connector 3.x 기본 5분)
정상 배포에서는 둘이 같은 토픽을 가리키니 그냥 이어 읽는다. 이번엔 두 출처가 어긋났다.
•
새 order source의 state → point 토픽
•
새 order source의 설정 → order 토픽
•
enumerator 입장에서 order 토픽의 파티션은 state에 없는 파티션 → 새로 발견한 것으로 보고 추가 할당
order source
├─ state 복원 → point 토픽
└─ 파티션 발견 → order 토픽
↓
두 토픽의 레코드가 order 싱크로 함께 나감
Plain Text
복사
state에 토픽명이 들어 있다는 점과 파티션 발견이 기본으로 켜져 있다는 점이 겹쳐서 남의 state가 붙은 source는 남의 토픽과 자기 토픽을 동시에 읽게되었다.
조치
모든 operator에 uid를 붙였다. 멀티 소스 잡은 키 이름이 uid 문자열에 들어가야 위치와 무관해진다.
src = env.from_source(...).uid(f"{key}-source")
out = src.process(...).uid(f"{key}-process")
ok = out.get_side_output(ok_tag).uid(f"{key}-success")
err = out.get_side_output(err_tag).uid(f"{key}-failed")
ok.sink_to(sink(key)).uid(f"{key}-sink")
err.sink_to(dlt_sink(key)).uid(f"{key}-dlt-sink")
Python
복사
uid를 붙인 줄은 6개인데 실제 operator는 8개
코드 | 생성되는 operator | uid |
from_source(…) | source | 직접 지정 |
.process(…) | process | 직접 지정 |
.get_side_output(…) × 2 | identity map 2개 — PyFlink가 자동 삽입 | 직접 지정 |
.sink_to(…) × 2 | Sink Writer 2개 | 직접 지정 |
— | Sink Committer 2개 | Writer uid에서 파생 |
•
get_side_output()은 Java와 달리 PyFlink에서 내부적으로 map(lambda i: i)을 삽입. 그래서 .get_side_output(...).uid(...)의 uid는 side output이 아니라 그 숨은 map에 붙음
•
key_by()도 _stream_key_by_map_operator를 만들므로 같은 처리가 필요
•
Committer는 Sink Committer: {Writer uid}로 파생되므로 따로 붙일 필요 없음
검증
uid를 붙인 뒤 키 하나를 맨 끝으로 옮겨 뒤 키 10개가 밀리게 하고 savepoint 모드로 재배포
지표 | 사고 때 | uid 적용 후 |
Skipping savepoint state | 마지막 1개 | 0 |
Discovered new partitions | 밀린 키 수만큼 | 0 |
Restoring state for N split(s) | 남의 split | 각 source가 자기 split 1개 |
vertex ID 변화 | 밀린 키 전부 | 0 / 92 |
_metadata의 operator ID 184개(23 × 8) 전부가 uid 해시와 일치하는 것까지 확인하고 마무리했다.
정리
•
Savepoint는 operator ID → state 맵. 복원은 ID 대조이고 이름·토픽·위치는 보지 않음
•
uid가 없으면 ID는 그래프 자리 번호. 자리가 밀리면 ID가 바뀌고, 자리가 같으면 남의 ID를 받음
•
Kafka source state에는 토픽명이 들어 있어 잘못 붙으면 남의 토픽을 읽고, 파티션 발견으로 자기 토픽도 읽어 섞임
•
PyFlink는 get_side_output, key_by에 숨은 operator를 만듦. 거기까지 붙여야 전부임
•
배포 성공과 state가 제 주인에게 붙은 것은 별개. Skipping savepoint state, Discovered new partitions, Restoring state for N split(s) 세 줄로 확인 필요