Kafka Connect 워커가 죽으면 같은 레코드가 두 번 들어갈까?

이번 시간에는 Kafka Connect가 offset을 언제 저장하는지를 분산 모드 워커와 FileStream 커넥터로 직접 돌려 보며 배워가는 시간입니다.

예를 들어서 주문 테이블이 바뀔 때마다 그 내용을 Kafka 토픽에 넣고, 토픽에 쌓인 주문을 다시 검색 엔진이나 데이터 웨어하우스(분석용 저장소)로 보내야 한다고 해볼게요.

직접 만들면 DB를 읽어 Kafka에 쓰는 프로그램과 Kafka를 읽어 바깥에 쓰는 프로그램이 하나씩 필요해요. 어디까지 읽었는지 저장하고, 재시작하면 그 위치부터 이어 읽고, 깨진 데이터를 만나면 멈출지 건너뛸지 정하는 코드도 함께 짜야 하죠.

Kafka Connect는 이 일을 프레임워크가 맡고, 사용자는 어떤 커넥터를 어떤 설정으로 돌릴지만 정하면 됩니다.

여기서 나오는 말 세 개를 먼저 정리할게요. 워커는 커넥터를 실행하는 Connect 프로세스이고, 커넥터는 실제로 데이터를 옮기는 일을 태스크로 나눠 워커 위에 띄워요. 컨버터는 레코드를 Kafka 메시지 바이트로 바꾸는 부품입니다.

다만 중복이 생겼을 때 원인을 찾으려면, 설정 뒤에서 offset을 언제 저장하는지 알아야 해요.

그래서 apache/kafka 이미지에 들어 있는 connect-distributed.sh로 워커를 띄워, 소스 offset이 저장되는 위치, 재시작 후 이어 읽기, offset 저장 전에 워커가 죽을 때 생기는 중복을 직접 확인해 볼게요.

실습 준비

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

• compose.yaml: Kafka 브로커, Connect 워커, PostgreSQL 컨테이너
• worker.properties: 분산 모드 Connect 워커 설정
• worker-eos.properties: exactly.once.source.support=enabled를 더한 워커 설정
• connectors/: REST API로 등록할 커넥터 설정 JSON(파일 2개)

compose.yaml로 컨테이너 세 개를 띄울게요. 이 글에서 쓰는 두 컨테이너는 이런 역할이에요.

• kafka: 브로커와 컨트롤러를 한 컨테이너에서 실행하는 KRaft 단일 노드예요. 컨테이너 안에서는 kafka:9092, 호스트에서는 localhost:9561로 접속합니다.
• connect: 같은 apache/kafka:4.3.1 이미지로 connect-distributed.sh를 실행하는 Connect 워커예요. REST API는 컨테이너 안 8083 포트이고, 호스트에서는 localhost:8561로 부릅니다.

postgres(PostgreSQL) 컨테이너도 함께 뜨지만 이 글에서는 쓰지 않아요.

아래는 워커 설정 파일 worker.properties예요.

bootstrap.servers=kafka:9092
group.id=connect-cluster
rest.advertised.host.name=connect

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
config.storage.replication.factor=1
offset.storage.replication.factor=1
status.storage.replication.factor=1

offset.flush.interval.ms=60000
plugin.path=/opt/kafka/libs/connect-file-4.3.1.jar,/opt/plugins

• config.storage.topic, offset.storage.topic, status.storage.topic: 분산 모드 워커가 쓰는 내부 토픽 이름이에요. 세 토픽의 복제 계수 기본값은 3이라서, 브로커가 한 대인 이 실습에서는 1로 낮췄습니다.
• offset.flush.interval.ms는 소스 태스크의 offset을 저장하는 주기예요. 기본값과 같은 60000(1분)을 일부러 적어 두었습니다.
• plugin.path는 커넥터, 컨버터, SMT(커넥터와 컨버터 사이에서 레코드를 하나씩 바꾸는 변환) 플러그인을 찾을 경로예요.

plugin.path에 connect-file-4.3.1.jar를 직접 넣은 데는 이유가 있어요.

FileStream 커넥터 jar는 이미지의 /opt/kafka/libs에 들어 있지만, 실행 스크립트(kafka-run-class.sh)가 클래스패스를 만들 때 정규식 connect-file.*\.jar로 이 jar를 빼거든요. 그래서 FileStream 커넥터를 쓰려면 plugin.path에 이 jar를 지정해 줘야 합니다.

compose 파일은 plugins/ 디렉터리를 워커의 /opt/plugins에 연결해요. 빈 디렉터리를 만들어 둘게요.

mkdir -p plugins

이제 실습에서 반복해 쓸 셸 함수를 만들어 놓을게요.

create는 connectors/<이름>.json 파일을 읽어 PUT /connectors/<이름>/config로 보내요. 이 API는 커넥터가 없으면 만들고, 있으면 설정을 바꿉니다. status는 상태 응답에서 커넥터와 태스크의 상태만 골라 보여 주는 함수예요.

아래 블록은 파일로 만들지 말고 실습 디렉터리에서 연 터미널에 그대로 붙여 넣어 주세요. 이후 명령은 모두 이 터미널에서 실행하고, 터미널을 새로 열면 다시 붙여 넣으시면 됩니다.

kafka() { docker compose exec -T kafka /opt/kafka/bin/"$@"; }

consume() {
  kafka kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic "$1" \
    --from-beginning --timeout-ms 5000 "${@:2}" 2>/dev/null
}

append() { docker compose exec -T connect sh -c "printf '%s\n' \"\$@\" >> $1" sh "${@:2}"; }

api() { curl -s "localhost:8561$1" "${@:2}"; }

create() {
  api "/connectors/$1/config" -X PUT -H 'Content-Type: application/json' --data @"connectors/$1.json" |
    jq -c '{name, tasks}'
}

status() { api "/connectors/$1/status" | jq -c '{connector: .connector.state, tasks: [.tasks[] | {id, state}]}'; }

그다음 컨테이너를 띄워 주세요.

docker compose up -d

실습 1: 소스 offset은 connect-offsets에 저장됩니다

첫 커넥터는 파일 한 줄을 메시지 하나로 보내는 FileStream 소스예요. 설정 파일 connectors/file-src.json은 이렇습니다.

{
  "connector.class": "FileStreamSource",
  "file": "/tmp/orders.txt",
  "topic": "orders-file",
  "tasks.max": "1"
}

워커 컨테이너 안에 주문 파일을 만들고 커넥터를 등록한 뒤, 상태와 토픽을 확인해 볼게요.

append /tmp/orders.txt 'O-1 created' 'O-2 created' 'O-3 created'
create file-src
status file-src
consume orders-file --formatter-property print.key=true

{"name":"file-src","tasks":[]}
{"connector":"RUNNING","tasks":[{"id":0,"state":"RUNNING"}]}
null	{"schema":{"type":"string","optional":false},"payload":"O-1 created"}
null	{"schema":{"type":"string","optional":false},"payload":"O-2 created"}
null	{"schema":{"type":"string","optional":false},"payload":"O-3 created"}

등록 응답의 tasks는 비어 있지만, 잠시 뒤 상태를 보면 커넥터와 태스크 0번이 모두 RUNNING이에요.

토픽에는 파일의 세 줄이 메시지 세 개로 들어갔어요. 키는 없고(null), 값은 schema와 payload로 감싼 JSON입니다. 이 모양은 워커에 설정한 JsonConverter 때문이에요. JsonConverter의 schemas.enable 기본값 true는 메시지마다 schema와 payload를 함께 써요.

소스 커넥터는 외부 시스템을 어디까지 읽었는지 스스로 정한 형식으로 Connect에 알려 줘요.

FileStream 소스는 파일 이름을 파티션으로, 파일 안의 바이트 위치를 offset으로 씁니다. 워커는 이 값을 모아 두었다가 offset.flush.interval.ms마다 connect-offsets에 써요.

커넥터를 만든 직후와, 저장 주기가 지난 뒤에 offset을 조회해 볼게요. GET /connectors/<이름>/offsets는 저장된 offset을 돌려줘요. 이어서 connect-offsets 토픽도 직접 읽습니다. 두 번째 조회는 1분쯤 지난 뒤에 해요.

api /connectors/file-src/offsets
api /connectors/file-src/offsets
consume connect-offsets --formatter-property print.key=true

{"offsets":[]}
{"offsets":[{"partition":{"filename":"/tmp/orders.txt"},"offset":{"position":36}}]}
["file-src",{"filename":"/tmp/orders.txt"}]	{"position":36}

처음에는 저장된 offset이 없어요. 메시지 세 개는 이미 토픽에 들어갔지만, 저장 주기가 오지 않았기 때문이에요.

1분 뒤에는 position이 36으로 저장됐어요. O-1 created와 줄바꿈이 12바이트이고 세 줄이니 36바이트입니다. connect-offsets 메시지의 키는 [커넥터 이름, 소스 파티션], 값은 소스 offset이에요.

이제 워커를 정상 종료로 재시작하고, 새 줄을 하나 추가한 뒤 토픽을 처음부터 읽어 볼게요.

docker compose restart connect
append /tmp/orders.txt 'O-4 created'
consume orders-file

{"schema":{"type":"string","optional":false},"payload":"O-1 created"}
{"schema":{"type":"string","optional":false},"payload":"O-2 created"}
{"schema":{"type":"string","optional":false},"payload":"O-3 created"}
{"schema":{"type":"string","optional":false},"payload":"O-4 created"}

재시작한 태스크는 저장된 위치 36부터 파일을 읽었어요. O-1부터 O-3은 다시 보내지 않았고, 새로 추가한 O-4만 보냈습니다.

정상 종료할 때 워커는 태스크를 멈추면서 그때까지의 offset을 저장하기 때문이에요.

실습 2: offset을 저장하기 전에 워커가 죽으면

앞에서 O-4를 보낸 뒤 아직 offset 저장 주기가 오지 않았어요.

여기서 두 줄을 더 쓰고, 메시지가 토픽에 들어간 것을 확인한 다음 워커를 SIGKILL로 끝내 볼게요. 그리고 다시 시작해서 토픽 전체를 읽습니다.

api /connectors/file-src/offsets
append /tmp/orders.txt 'O-5 created' 'O-6 created'
consume orders-file | tail -2
api /connectors/file-src/offsets
docker compose kill -s SIGKILL connect
docker compose start connect
consume orders-file

{"offsets":[{"partition":{"filename":"/tmp/orders.txt"},"offset":{"position":36}}]}
{"schema":{"type":"string","optional":false},"payload":"O-5 created"}
{"schema":{"type":"string","optional":false},"payload":"O-6 created"}
{"offsets":[{"partition":{"filename":"/tmp/orders.txt"},"offset":{"position":36}}]}
{"schema":{"type":"string","optional":false},"payload":"O-1 created"}
{"schema":{"type":"string","optional":false},"payload":"O-2 created"}
{"schema":{"type":"string","optional":false},"payload":"O-3 created"}
{"schema":{"type":"string","optional":false},"payload":"O-4 created"}
{"schema":{"type":"string","optional":false},"payload":"O-5 created"}
{"schema":{"type":"string","optional":false},"payload":"O-6 created"}
{"schema":{"type":"string","optional":false},"payload":"O-4 created"}
{"schema":{"type":"string","optional":false},"payload":"O-5 created"}
{"schema":{"type":"string","optional":false},"payload":"O-6 created"}

출력을 차례로 보면 이렇습니다.

1. 강제 종료 직전, O-5와 O-6은 이미 토픽에 들어가 있었어요. 그런데 저장된 offset은 여전히 36이에요.
2. 재시작한 태스크는 36부터 다시 읽었어요. 그 뒤에 있는 O-4, O-5, O-6을 한 번 더 보냈습니다.
3. 토픽에는 O-4부터 O-6까지가 두 번씩 남았어요. O-6 뒤에 다시 O-4가 나오는 부분이 재시작 후 보낸 메시지입니다.

왜 이런 일이 생길까요? Kafka에 메시지를 쓰는 일과 소스 offset을 저장하는 일이 따로 일어나기 때문이에요.

그 사이에 워커가 죽으면, 재시작한 태스크는 이미 보낸 데이터를 다시 보냅니다. 최대 offset.flush.interval.ms만큼의 데이터가 중복될 수 있어요.

컨슈머가 commit 전에 죽으면 같은 메시지를 다시 읽는 것과 같은 구조죠. 그러니 이 토픽을 읽는 쪽은 같은 메시지를 두 번 받아도 결과가 같도록 만들어야 해요.

커넥터가 지원한다면 Connect 쪽에서 이 중복을 막는 설정도 있어요.

실습 3: exactly.once.source.support로 소스 중복 막기

Kafka 3.3부터 분산 모드 워커는 소스 커넥터의 메시지와 소스 offset을 한 트랜잭션으로 쓰는 기능을 제공해요. 공식 문서 기준으로 정리하면 이렇습니다.

• 워커 설정 exactly.once.source.support: 기본값은 disabled예요. 새 클러스터는 모든 워커에 enabled를 넣고, 이미 운영 중인 클러스터는 preparing으로 한 번, enabled로 한 번 순차 재시작해요. standalone 모드에서는 쓸 수 없습니다.
• 커넥터 설정 exactly.once.support: required로 두면 커넥터가 exactly-once를 지원하는지 생성할 때 미리 검사하고, 지원하지 않으면 만들지 않아요. 기본값 requested는 검사하지 않습니다.
• 커넥터 설정 transaction.boundary: 기본값 poll은 태스크가 한 번 넘겨준 레코드 묶음마다 트랜잭션을 하나 커밋해요.

앞에서 만든 file-src를 지우고, 워커 설정에 exactly.once.source.support=enabled 한 줄만 추가한 worker-eos.properties로 워커를 다시 띄울게요.

그리고 exactly.once.support를 required로 둔 FileStream 소스 connectors/eos-src.json으로 앞의 강제 종료 실험을 반복해 봅니다. 워커 컨테이너를 새로 만들기 때문에 파일도 새 파일 /tmp/eos.txt를 써요.

{
  "connector.class": "FileStreamSource",
  "file": "/tmp/eos.txt",
  "topic": "orders-eos",
  "tasks.max": "1",
  "exactly.once.support": "required"
}

api /connectors/file-src -X DELETE

WORKER_CONF=worker-eos.properties docker compose up -d connect

append /tmp/eos.txt 'E-1 created' 'E-2 created'
create eos-src
append /tmp/eos.txt 'E-3 created' 'E-4 created'
api /connectors/eos-src/offsets
kafka kafka-transactions.sh --bootstrap-server kafka:9092 list

docker compose kill -s SIGKILL connect
docker compose start connect

consume orders-eos --isolation-level read_committed
api /connectors/eos-src/offsets

{"name":"eos-src","tasks":[]}
{"offsets":[{"partition":{"filename":"/tmp/eos.txt"},"offset":{"position":48}}]}
TransactionalId                	Coordinator	ProducerId	TransactionState
connect-cluster-connect-cluster	1          	1         	CompleteCommit
connect-cluster-eos-src-0      	1          	2         	CompleteCommit
{"schema":{"type":"string","optional":false},"payload":"E-1 created"}
{"schema":{"type":"string","optional":false},"payload":"E-2 created"}
{"schema":{"type":"string","optional":false},"payload":"E-3 created"}
{"schema":{"type":"string","optional":false},"payload":"E-4 created"}
{"offsets":[{"partition":{"filename":"/tmp/eos.txt"},"offset":{"position":48}}]}

이번에는 네 줄을 보낸 지 몇 초 만에 offset이 48(네 줄)로 저장돼 있어요. 1분 주기를 기다리지 않고, 트랜잭션을 커밋할 때 offset도 함께 저장했기 때문이에요.

태스크용 트랜잭션 프로듀서(connect-cluster-eos-src-0)도 보이죠. ProducerId 숫자는 그 전에 브로커에 접속한 프로듀서에 따라 다를 수 있어요. 강제 종료 후 재시작해도 E-1부터 E-4는 한 번씩만 있습니다.

이 결과를 얻으려면 커넥터가 이 기능을 지원해야 하고, 토픽을 읽는 쪽이 커밋된 트랜잭션의 메시지만 읽는 read_committed로 읽어야 해요. 트랜잭션과 read_committed의 동작이 더 궁금하면 exactly-once를 다룬 글을 참고하세요.

정리

범위: apache/kafka:4.3.1의 connect-distributed.sh 분산 모드 워커 1대, 내장 FileStream 커넥터.

• offset 저장 주기(1분) 전에 워커를 SIGKILL로 끝내자 재시작한 태스크가 저장된 위치 36부터 다시 읽어 O-4부터 O-6까지가 두 번씩 남았습니다. 소스 커넥터 토픽을 읽는 쪽은 최대 offset.flush.interval.ms만큼의 중복을 견디게 만들어 주세요.
• exactly.once.source.support=enabled 워커와 exactly.once.support=required 커넥터로 같은 강제 종료를 반복하자 offset이 몇 초 만에 48로 저장됐고 E-1부터 E-4가 한 번씩만 남았습니다. 이 기능을 쓸 때는 읽는 쪽을 read_committed로 읽게 하세요.
• 이 글에 싣지 않은 추가 측정에서는 싱크의 errors.tolerance 기본값 none에서 JSON이 아닌 메시지 한 건에 태스크가 FAILED가 됐고, errors.tolerance=all과 DLQ(처리하지 못한 레코드를 따로 모으는 토픽)를 켜자 나머지는 파일에 써지고 그 한 건은 원본 위치와 오류 헤더를 달고 DLQ 토픽에 들어갔습니다. all을 쓸 때는 DLQ와 오류 헤더를 함께 켜 주세요.