Hong로그인
# Flink# Kafka

Flink 잡이 재시작하면 Kafka 결과는 왜 두 번 들어갈까?

@Hong2026-10-14

이 글에 담긴 코드와 결과는 직접 실행해 측정한 값을 바탕으로 작성했습니다. 실측은 Hong과 팀원들이 진행했고, 그 결과를 바탕으로 AI가 글을 쓰고 Hong이 검토합니다.

실습 코드 내려받기

질문
Flink 잡이 장애 뒤 재시작하면 Kafka 출력이 왜 중복될까?
답
재시작한 잡은 마지막 체크포인트의 offset부터 다시 처리해 at-least-once 싱크에서는 1200건 중 100~120건이 중복됐고, exactly-once 싱크를 read_committed로 읽으면 1200건이 한 번씩이었다.
조건
Flink 2.2.1(flink:2.2.1-java17), Kafka 4.3.1, Kafka SQL 커넥터 5.0.0-2.2, 병렬도 1, 입력 1200건, TaskManager 강제 종료

환경

Flink
2.2.1(flink:2.2.1-java17), JobManager와 TaskManager 하나씩, 병렬도 1
Kafka
4.3.1(apache/kafka:4.3.1), 브로커 하나
커넥터
Kafka SQL 커넥터 5.0.0-2.2
실행
Docker Compose, 체크포인트 간격 20초
데이터
입력 1200건, 토픽 파티션 1개
장애
TaskManager 컨테이너 강제 종료(SIGKILL)
목차
  1. 실습 준비
  2. 실습 1: at-least-once 싱크에서 재시작하면
  3. 실습 2: exactly-once 싱크로 같은 장애를
  4. 정리

이번 시간에는 Docker로 띄운 Flink와 Kafka에서 TaskManager를 강제로 종료해 보면서, 체크포인트와 싱크의 전달 보장이 재시작 뒤의 출력을 어떻게 바꾸는지 측정하며 배워가는 시간입니다.

예를 들어서 우리가 Kafka 토픽 ck-in에서 {"seq":1}부터 {"seq":1200}까지 1200건을 읽어 그대로 ck-out-alo 토픽에 쓰는 Flink 잡을 돌린다고 해볼게요. 체크포인트는 20초마다 성공하고 있었어요.

입력을 흘리는 도중에 TaskManager를 강제로 종료하면 잡은 혼자 복구하는데, 출력 토픽에는 일부 seq가 두 번씩 들어 있어요. 왜 그럴까요?

체크포인트를 켰다고 해서 출력이 한 번만 나가는 건 아니기 때문이에요. 장애가 나면 Flink는 마지막으로 성공한 체크포인트로 돌아가 그 뒤 구간을 다시 처리하고, 싱크가 이미 쓴 결과는 그대로 남습니다.

실습 준비

이 글의 명령은 모두 실습 코드의 flink-checkpoint-exactly-once/ 디렉터리에서 실행합니다.

  • compose.yaml: Kafka 브로커 하나와 Flink JobManager, TaskManager
  • common.sh: 토픽 초기화, 입력 보내기, 잡 제출, 로그·상태 조회, TaskManager 강제 종료, 결과 세기 셸 함수
  • sql/: 잡을 만드는 Flink SQL 파일 2개(alo.sql, eos.sql)
  • crash.sh: 입력을 흘리는 중에 TaskManager를 강제 종료하고 결과를 세는 스크립트

구성은 Kafka 브로커 하나와, 잡을 관리하는 Flink JobManager, 실제로 데이터를 처리하는 TaskManager 하나씩이에요. 잡은 Flink SQL 클라이언트로 제출하고, Kafka SQL 커넥터 jar는 두 Flink 컨테이너에 마운트합니다.

Flink 이미지에는 Kafka 커넥터가 들어 있지 않아요. 그래서 Maven Central에서 Flink 2.2용 커넥터를 받아 오고, 체크포인트를 저장할 ckpt 디렉터리도 같이 만들어 줄게요.

mkdir -p lib sql ckpt
curl -sSfL -o lib/flink-sql-connector-kafka-5.0.0-2.2.jar \
  https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/5.0.0-2.2/flink-sql-connector-kafka-5.0.0-2.2.jar
ls lib

flink-sql-connector-kafka-5.0.0-2.2.jar

compose.yaml은 apache/kafka:4.3.1(메모리 512MB, CPU 0.5), flink:2.2.1-java17 JobManager(1,024MB, CPU 0.5)와 TaskManager(1,280MB, CPU 1.5)를 띄웁니다.

  • execution.checkpointing.dir: file:///ckpt: 체크포인트를 저장할 경로예요. JobManager와 TaskManager가 같은 경로를 봐야 복구할 수 있어서, 호스트의 ./ckpt를 두 컨테이너의 /ckpt에 똑같이 마운트했습니다.
  • taskmanager.numberOfTaskSlots: 1: TaskManager 하나가 동시에 돌릴 수 있는 슬롯 수예요. 잡 하나를 병렬도 1로 돌리니까 1개면 충분합니다.
  • JobManager의 8081 포트(Web UI와 REST API)는 호스트의 7660으로 엽니다.

compose 프로젝트 이름을 flkcp로 정해서 띄운 다음, 클러스터가 슬롯 1개를 인식했는지 REST API로 확인해 볼게요. 응답 JSON에 "taskmanagers":1, "slots-total":1이 보이면 준비가 끝난 거예요.

export COMPOSE_PROJECT_NAME=flkcp
docker compose up -d
curl -s localhost:7660/overview; echo

crash.sh는 common.sh의 셸 함수로 실습 순서를 묶은 스크립트예요. 결과를 읽는 데 필요한 함수만 보면 이렇습니다.

  • reset_topics: 실행 중인 잡을 모두 취소하고, 실습 토픽 세 개(ck-in, ck-out-alo, ck-out-eos)를 지운 뒤 파티션 1개로 다시 만들어요. 파티션을 하나로 두면 seq 순서와 offset이 그대로 대응해서 결과를 읽기 쉽습니다.
  • feed <시작> <끝> <간격>: {"seq":N}을 정해진 간격으로 하나씩 ck-in에 보내요.
  • submit <SQL 파일>: 실행 중인 잡을 취소하고 SQL 파일을 SQL 클라이언트로 제출해요.
  • ckpt: 체크포인트 통계(완료·실패·복구 횟수, 마지막 체크포인트 번호)를 REST API로 읽어요.
  • wait_ckpt, wait_restored: 다음 체크포인트 완료, 체크포인트에서 복구를 기다리고 그 시간을 찍습니다. Flink 로그와 맞추려고 시간은 모두 UTC예요.
  • kill_tm: TaskManager 컨테이너를 강제로 종료(SIGKILL)하고 바로 새로 띄웁니다.
  • count <토픽> <isolation.level>: 토픽을 처음부터 읽어 총 줄 수, 서로 다른 seq 수, 두 번 이상 나온 seq 구간을 세요.

crash.sh <SQL 파일> <출력 토픽>은 입력 1200건을 0.1초 간격으로 흘리는 중에, 체크포인트가 두 번 끝나고 10초가 지난 시점에 TaskManager를 종료해요. 10초를 기다리는 건 체크포인트 이후에 처리한 구간을 만들어 두기 위해서예요.

TaskManager를 종료한 직후에는 출력 토픽의 끝 offset을 읽습니다. 이때는 아무도 출력 토픽에 쓰지 않으니까, 이 값이 장애 전까지 쓴 결과의 수라고 보시면 됩니다. 복구 뒤 입력이 끝나면 체크포인트 두 번을 더 기다렸다가 출력 토픽을 셉니다.

실습 1: at-least-once 싱크에서 재시작하면

잡은 sql/alo.sql로 만들어요. 체크포인트 간격을 20초로 켜고, ck-in을 읽는 ck_in 테이블과 ck-out-alo에 쓰는 ck_out 테이블을 만든 뒤 seq를 그대로 넘기는 잡입니다.

  • scan.startup.mode = 'earliest-offset': 잡을 처음 시작할 때 토픽의 처음부터 읽어요. 체크포인트에서 복구할 때는 이 값이 아니라 체크포인트에 저장된 offset을 씁니다.
  • properties.group.id = 'flkcp': Kafka 소스가 offset을 커밋할 컨슈머 그룹 이름이에요.
  • sink.delivery-guarantee: 싱크의 전달 보장입니다. none, at-least-once, exactly-once 중 하나이고 Kafka SQL 커넥터의 기본값은 at-least-once예요. 여기서는 기본값을 명시만 했습니다.
bash crash.sh alo.sql ck-out-alo

취소: f8d52bba44763384c5588e86c9da2ba3
Created topic ck-in.
Created topic ck-out-alo.
Created topic ck-out-eos.
Job ID: 7ef9c0c12079649811ec7d0f6b7d27c8
체크포인트 완료: 15:17:05
체크포인트 완료: 15:17:24
GROUP TOPIC CURRENT-OFFSET LOG-END-OFFSET LAG
flkcp ck-in 255 305 50
TaskManager 종료: 15:17:34
ck-out-alo:0:355
복구됨: 15:17:55
{"completed":2,"failed":1,"restored":1,"latest":2,"restored_from":2}
15:16:54,119 Job insert-into_default_catalog.default_database.ck_out (7ef9c0c12079649811ec7d0f6b7d27c8) switched from state CREATED to RUNNING.
15:17:54,035 TaskManager with id 172.29.0.4:33923-0159e5 is no longer reachable.
15:17:54,035 Closing TaskExecutor connection 172.29.0.4:33923-0159e5 because: TaskManager with id 172.29.0.4:33923-0159e5 is no longer reachable.
15:17:54,212 Job insert-into_default_catalog.default_database.ck_out (7ef9c0c12079649811ec7d0f6b7d27c8) switched from state RUNNING to RESTARTING.
15:17:54,215 Disconnect TaskExecutor 172.29.0.4:33923-0159e5 because: TaskManager with id 172.29.0.4:33923-0159e5 is no longer reachable.
15:17:55,313 Job insert-into_default_catalog.default_database.ck_out (7ef9c0c12079649811ec7d0f6b7d27c8) switched from state RESTARTING to RUNNING.
15:17:55,314 Restoring job 7ef9c0c12079649811ec7d0f6b7d27c8 from Checkpoint 2 @ 1791645443650 for 7ef9c0c12079649811ec7d0f6b7d27c8 located at file:/ckpt/7ef9c0c12079649811ec7d0f6b7d27c8/chk-2.
15:16:54,589 Adding split(s) to reader: [[Partition: ck-in-0, StartingOffset: -2, StoppingOffset: -9223372036854775808]]
15:17:55,870 Adding split(s) to reader: [[Partition: ck-in-0, StartingOffset: 255, StoppingOffset: -9223372036854775808]]
입력 끝: 15:19:08
체크포인트 완료: 15:19:11
체크포인트 완료: 15:19:31
{"completed":7,"failed":1,"restored":1,"latest":8,"restored_from":2}
총 1300줄, 서로 다른 seq 1200개(가장 큰 seq 1200), 두 번 이상 나온 seq 100개 (256~355)
GROUP TOPIC CURRENT-OFFSET LOG-END-OFFSET LAG
flkcp ck-in 1200 1200 0

출력을 순서대로 따라가 볼게요.

  1. 15:17:24에 체크포인트 2가 끝났어요. 이 체크포인트에는 "입력 offset 255부터 읽으면 된다"가 저장됐습니다. seq는 1부터, offset은 0부터 시작하니까 seq 255까지 처리했다는 뜻이에요.
  2. 10초 뒤인 15:17:34에 TaskManager를 종료했습니다. 그때까지 출력 토픽에는 355건이 쓰였어요. seq 1부터 355까지예요.
  3. 15:17:55에 체크포인트 2에서 복구했고, TaskManager 로그의 StartingOffset: 255처럼 소스는 offset 255부터 다시 읽었습니다. 처음 시작할 때의 -2는 "가장 처음부터"(earliest-offset)를 뜻하는 값이에요.
  4. 그래서 seq 256부터 355까지 100건이 다시 처리되어 출력 토픽에 한 번 더 쓰였습니다. 마지막 집계의 중복 구간 256~355와 같죠.

빠진 seq는 없어요. at-least-once 싱크는 이름대로 유실은 막지만 중복은 막지 않아요.

같은 실습을 세 번 했고, 중복은 100건, 109건, 120건이었어요. 중복 건수는 체크포인트가 끝난 뒤 얼마나 지나 장애가 났는지, 그 사이 입력이 얼마나 들어왔는지로 정해져요. 그러니 체크포인트 간격이 길수록 최악의 경우 중복 구간도 길어집니다.

컨슈머 그룹 flkcp의 CURRENT-OFFSET도 255였지만, Flink Kafka 소스는 복구할 때 이 값이 아니라 체크포인트에 저장된 offset을 씁니다. 그룹에 커밋한 offset은 lag 같은 진행 상황을 보여 주는 용도예요.

실습 2: exactly-once 싱크로 같은 장애를

출력 토픽의 중복을 막으려면, 체크포인트 이후에 쓴 결과를 장애가 났을 때 없던 일로 만들 수 있어야 해요. Kafka 싱크의 exactly-once 모드는 이걸 Kafka 트랜잭션으로 합니다.

  • 체크포인트와 체크포인트 사이에 쓰는 결과는 하나의 트랜잭션에 담아요. 이 트랜잭션은 아직 커밋하지 않습니다.
  • 체크포인트가 성공하면 그 트랜잭션을 커밋해요. 이때부터 read_committed 컨슈머가 결과를 봅니다.
  • 장애가 나서 체크포인트로 돌아가면, 커밋하지 않은 트랜잭션은 중단돼요. 다시 처리한 결과는 새 트랜잭션에 쓰입니다.

중단된 트랜잭션의 메시지도 토픽에서 지워지지는 않아요. 컨슈머가 read_committed로 읽어야 건너뜁니다.

sql/eos.sql은 출력 토픽을 ck-out-eos로 두고 싱크 테이블에 아래 세 옵션을 넣은 파일이에요. 소스 테이블과 체크포인트 간격은 alo.sql과 같아요.

  • sink.delivery-guarantee = 'exactly-once': 결과를 트랜잭션으로 쓰고 체크포인트에서 커밋해요. 체크포인트가 켜져 있어야 합니다.
  • sink.transactional-id-prefix: Flink가 만드는 transactional.id의 앞부분이에요. exactly-once에서는 필수이고, 같은 Kafka 클러스터를 쓰는 다른 잡과 겹치지 않아야 합니다.
  • properties.transaction.timeout.ms: properties.로 시작하는 옵션은 Kafka 프로듀서 설정으로 그대로 넘어가요. 트랜잭션 타임아웃을 5분으로 줬습니다.

출력 토픽이 ck-out-eos이면 crash.sh는 TaskManager를 종료한 직후와 복구 뒤에 kafka-transactions.sh로 flkcp-eos로 시작하는 트랜잭션 목록을 보고, Ongoing(진행 중)인 것은 describe로 자세히 봐요. 마지막에는 출력 토픽을 read_uncommitted, read_committed, 그리고 --isolation-level 없이 각각 읽습니다.

bash crash.sh eos.sql ck-out-eos

취소: c02df88771380a4de49c88aeea642e3e
Created topic ck-in.
Created topic ck-out-alo.
Created topic ck-out-eos.
Job ID: 5dce493b812752371247788f0dc379c7
체크포인트 완료: 15:42:25
체크포인트 완료: 15:42:44
GROUP TOPIC CURRENT-OFFSET LOG-END-OFFSET LAG
flkcp ck-in 226 266 40
TaskManager 종료: 15:42:54
== 종료 직후 트랜잭션
TransactionalId	Coordinator	ProducerId	TransactionState	
flkcp-eos-0-3  	1          	47        	Ongoing         	
flkcp-eos-0-2  	1          	46        	CompleteCommit  	
flkcp-eos-0-1  	1          	43        	Empty           	
CoordinatorId	TransactionalId	ProducerId	ProducerEpoch	TransactionState	TransactionTimeoutMs	CurrentTransactionStartTimeMs	TransactionDurationMs	TopicPartitions	
1            	flkcp-eos-0-3  	47        	18           	Ongoing         	300000              	1791646964901                	26248                	ck-out-eos-0   	
ck-out-eos:0:328
복구됨: 15:43:16
{"completed":2,"failed":1,"restored":1,"latest":2,"restored_from":2}
15:42:17,539 Job insert-into_default_catalog.default_database.ck_out (5dce493b812752371247788f0dc379c7) switched from state CREATED to RUNNING.
15:43:07,636 TaskManager with id 172.29.0.4:40287-24247d is no longer reachable.
15:43:07,640 Job insert-into_default_catalog.default_database.ck_out (5dce493b812752371247788f0dc379c7) switched from state RUNNING to RESTARTING.
15:43:07,641 Disconnect TaskExecutor 172.29.0.4:40287-24247d because: TaskManager with id 172.29.0.4:40287-24247d is no longer reachable.
15:43:08,668 Job insert-into_default_catalog.default_database.ck_out (5dce493b812752371247788f0dc379c7) switched from state RESTARTING to RUNNING.
15:43:08,669 Restoring job 5dce493b812752371247788f0dc379c7 from Checkpoint 2 @ 1791646963961 for 5dce493b812752371247788f0dc379c7 located at file:/ckpt/5dce493b812752371247788f0dc379c7/chk-2.
15:43:14,641 TaskManager with id 172.29.0.4:40287-24247d is no longer reachable.
15:43:14,642 Closing TaskExecutor connection 172.29.0.4:40287-24247d because: TaskManager with id 172.29.0.4:40287-24247d is no longer reachable.
15:42:18,278 Adding split(s) to reader: [[Partition: ck-in-0, StartingOffset: -2, StoppingOffset: -9223372036854775808]]
15:43:16,272 Adding split(s) to reader: [[Partition: ck-in-0, StartingOffset: 226, StoppingOffset: -9223372036854775808]]
입력 끝: 15:44:32
체크포인트 완료: 15:44:48
체크포인트 완료: 15:45:07
{"completed":8,"failed":1,"restored":1,"latest":9,"restored_from":2}
== 복구 뒤 트랜잭션
TransactionalId	Coordinator	ProducerId	TransactionState	
flkcp-eos-0-8  	1          	54        	CompleteCommit  	
flkcp-eos-0-7  	1          	53        	CompleteCommit  	
flkcp-eos-0-6  	1          	52        	CompleteCommit  	
flkcp-eos-0-5  	1          	51        	CompleteCommit  	
flkcp-eos-0-3  	1          	47        	CompleteCommit  	
flkcp-eos-0-2  	1          	46        	CompleteCommit  	
flkcp-eos-0-1  	1          	43        	Empty           	
CoordinatorId	TransactionalId	ProducerId	ProducerEpoch	TransactionState	TransactionTimeoutMs	CurrentTransactionStartTimeMs	TransactionDurationMs	TopicPartitions	
1            	flkcp-eos-0-3  	47        	22           	CompleteCommit  	300000              	1791646996403                	120644               	               	
== TaskManager 로그의 트랜잭션 오류
15:43:15,752 [Producer clientId=producer-flkcp-eos-0-2, transactionalId=flkcp-eos-0-2] Transiting to fatal error state due to org.apache.flink.kafka.shaded.org.apache.kafka.common.errors.ProducerFencedException: There is a newer producer with the same transactionalId which fences the current one.
== read_uncommitted
총 1300줄, 서로 다른 seq 1200개(가장 큰 seq 1200), 두 번 이상 나온 seq 100개 (227~326)
== read_committed
총 1200줄, 서로 다른 seq 1200개(가장 큰 seq 1200), 두 번 이상 나온 seq 0개
== --isolation-level 없이
    1300
GROUP TOPIC CURRENT-OFFSET LOG-END-OFFSET LAG
flkcp ck-in 1200 1200 0

같은 브로커에서 실습을 여러 번 반복하다 보니 트랜잭션 목록에 상태가 Empty인 id가 많이 나와서, 이번 실행의 flkcp-eos-0-1만 남기고 나머지 Empty 줄은 지우고 실었어요. 새 브로커에서 처음 실행하시면 목록이 더 짧고, ProducerEpoch 값도 이보다 작게 나올 거예요.

  1. 체크포인트 2가 15:42:44에 끝났고, 입력 offset 226을 저장했어요.
  2. 15:42:54에 TaskManager를 종료한 직후, 체크포인트 2 뒤에 열린 flkcp-eos-0-3 트랜잭션이 Ongoing입니다. 체크포인트 2까지의 결과를 담은 flkcp-eos-0-2는 이미 CompleteCommit이에요.
  3. 15:43:08에 체크포인트 2에서 복구했고 소스는 offset 226부터 다시 읽었습니다. at-least-once와 같은 재처리예요.
  4. 복구 뒤 flkcp-eos-0-3은 ProducerEpoch가 18에서 22로 올라갔고 상태는 CompleteCommit입니다. 복구한 잡이 같은 id를 다시 써서 새 트랜잭션을 열고 커밋한 거예요. 장애 전에 이 id로 쓴 100건은 커밋되지 않은 채 남았습니다.

read_uncommitted로 읽으면 1300건에 seq 227부터 326까지 100건이 두 번 나와요. 반면 read_committed로 읽으면 1200건에 중복이 없습니다. 싱크는 똑같이 두 번 썼지만, 처음 쓴 100건은 중단된 트랜잭션에 들어 있어서 read_committed 컨슈머가 건너뛴 거예요.

--isolation-level 없이 읽은 콘솔 컨슈머는 1300건을 읽었어요. 기본값이 read_uncommitted라는 뜻이죠. 그러니 출력 토픽을 읽는 쪽에 isolation.level=read_committed를 직접 줘야 합니다. 주지 않으면 exactly-once 싱크를 켜도 중복이 그대로 보여요.

TaskManager 로그의 ProducerFencedException 한 줄은 복구 중에 이전 프로듀서가 같은 id의 새 프로듀서에게 밀려나면서 남은 거예요. 잡은 이 예외로 실패하지 않았습니다.

같은 실습을 네 번 했는데, read_uncommitted의 중복은 100건에서 118건 사이였고 read_committed는 네 번 모두 1200건이 한 번씩이었어요.

정리

범위: Flink 2.2.1(flink:2.2.1-java17), Kafka 4.3.1, Kafka SQL 커넥터 5.0.0-2.2, 병렬도 1, 입력 1200건, 장애는 TaskManager 강제 종료.

  • at-least-once 싱크에서는 복구된 offset 다음부터 장애 직전까지가 다시 쓰여 세 번 반복에 100건, 109건, 120건이 중복됐고 빠진 seq는 없었습니다. 체크포인트 간격이 길수록 최악의 중복 구간도 길어져요.
  • 복구 기준은 체크포인트에 저장된 offset입니다. 컨슈머 그룹에 커밋된 offset은 진행 상황 모니터링에만 쓰세요.
  • exactly-once 싱크에서 read_committed로 읽은 결과는 네 번 모두 1200건이 한 번씩이었지만, read_uncommitted와 --isolation-level 없는 콘솔 컨슈머는 1300건에 100건이 중복됐습니다. 출력 토픽을 읽는 모든 컨슈머에 isolation.level=read_committed를 직접 주세요.
  • 이 글에 싣지 않은 추가 측정(입력 600건)에서는 트랜잭션 타임아웃 10초를 체크포인트 간격 20초보다 짧게 두자 잡은 계속 RUNNING이었는데 네 번 중 세 번 결과가 사라졌습니다(8건, 372건, 39건). 트랜잭션 타임아웃은 체크포인트 간격보다 짧게 잡지 마세요.
내 링크 만들기