Kafka Connect

Connect가 해결하는 문제

데이터베이스 변경을 Kafka로 보내고, Kafka의 이벤트를 검색 엔진과 데이터 웨어하우스로 내려보내는 코드는 어느 조직에나 있습니다. 그리고 그 코드는 거의 항상 같은 문제를 다시 풉니다 — 어디까지 읽었는지 기록하고, 실패하면 재시도하고, 프로세스가 죽으면 다른 노드에서 이어받고, 스키마를 변환합니다. 공식 문서는 Connect의 목적을 "Kafka와 다른 시스템 사이에서 데이터를 확장성 있고 신뢰성 있게 흘려보내는 도구"로 정의하며, 그 특징으로 커넥터 공통 프레임워크, standalone/distributed 두 모드, REST 인터페이스, 자동 오프셋 관리, 기본 제공되는 분산·확장성, 스트리밍/배치 통합을 듭니다.

Connect를 쓰는 판단 기준은 간단합니다. 데이터를 옮기기만 한다면 Connect, 옮기면서 계산해야 한다면 Kafka Streams입니다. 이 경계는 뒤에서 다루는 SMT의 제약에서 다시 등장합니다.

다섯 개의 명사

standalone과 distributed 모드

Connect는 두 가지 실행 모드를 지원합니다. standalone은 모든 작업이 단일 프로세스에서 수행되며 설정이 간단하지만, 공식 문서는 내고장성(fault tolerance) 같은 Connect의 일부 기능을 누리지 못한다고 명시합니다. distributed는 작업 자동 분배, 동적 스케일 인/아웃, 그리고 활성 태스크와 설정·오프셋 커밋 데이터 양쪽에 대한 내고장성을 제공합니다.

모든 워커가 요구하는 설정

# 모든 워커 공통 (standalone/distributed 동일)
bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
plugin.path=/opt/kafka/connect-plugins

# distributed 모드 전용 — 이 클러스터를 식별하는 값. 컨슈머 그룹 ID와 겹치면 안 됩니다
group.id=connect-prod

# 내부 토픽 3종. 사전에 직접 만드는 것이 권장됩니다
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status

# 브로커 3대 이상이면 3을 씁니다 (기본값도 3)
config.storage.replication.factor=3
offset.storage.replication.factor=3
status.storage.replication.factor=3
offset.storage.partitions=25
status.storage.partitions=5

# REST 리스너 (기본값 http://:8083)
listeners=http://0.0.0.0:8083

내부 토픽 3종

distributed 모드의 Connect 클러스터는 상태를 워커 로컬에 두지 않고 Kafka 토픽 세 개에 둡니다. 그래서 워커가 죽어도 다른 워커가 이어받을 수 있습니다. 공식 문서는 이 세 토픽을 Connect 기동 전에 직접 만들 것을 권장합니다. 자동 생성에 맡기면 브로커의 기본 파티션 수·복제 계수가 적용되어 용도에 맞지 않을 수 있기 때문입니다.

커넥터·태스크의 상태 변경은 status.storage.topic에 발행되고 모든 워커가 이 토픽을 구독합니다. 워커가 이 토픽을 비동기로 읽기 때문에 상태 변경이 REST API에 보이기까지 약간의 지연이 있습니다. 커넥터를 만든 직후 RUNNING이 아니라 UNASSIGNED가 보이는 것은 정상입니다.

source와 sink의 오프셋 저장 위치는 다릅니다

이 장에서 하나만 기억해야 한다면 이것입니다. 같은 "오프셋"이라는 단어를 쓰지만 source와 sink는 완전히 다른 것을 다른 곳에 저장합니다.

// FileStreamSourceConnector — partition/offset 구조를 커넥터가 정의합니다
{
  "offsets": [
    { "partition": { "filename": "test.txt" },
      "offset":    { "position": 30 } }
  ]
}

// FileStreamSinkConnector — 모든 sink 커넥터가 이 공통 형식을 씁니다
{
  "offsets": [
    { "partition": { "kafka_topic": "test", "kafka_partition": 0 },
      "offset":    { "kafka_offset": 5 } },
    { "partition": { "kafka_topic": "test", "kafka_partition": 1 },
      "offset":    null }
  ]
}

offset 필드가 null이면 해당 파티션의 오프셋을 초기화합니다. 공식 문서는 source 커넥터의 요청 본문 형식은 커넥터 구현에 따라 다르고, sink 커넥터는 공통 형식이라는 점을 명시합니다 — 이것도 위 비대칭의 결과입니다.

source 오프셋 커밋 주기

Converter — Kafka에 무슨 바이트가 쓰이는가

커넥터는 데이터를 Connect의 내부 표현(Struct/Map + Schema)으로 다룹니다. 그것을 Kafka에 저장될 바이트열로 바꾸는 것이 converter입니다. converter는 커넥터와 독립이라서, 같은 JDBC source 커넥터를 Avro로도 JSON으로도 쓸 수 있습니다.

value.converter=org.apache.kafka.connect.json.JsonConverter
# schemas.enable 을 지정하지 않음 → 기본값 true

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

SMT — Single Message Transform

공식 문서는 transform을 "가벼운 메시지 단위(message-at-a-time) 수정"이라고 정의하며, 데이터 정리와 이벤트 라우팅에 편리하다고 설명합니다. 이 정의에 SMT의 능력과 한계가 모두 들어 있습니다.

기본 제공 SMT

SMT 클래스 이름에 $Key / $Value 접미어를 붙여 키에 적용할지 값에 적용할지를 고릅니다. 예를 들어 InsertField$Value는 값에, ExtractField$Key는 키에 적용됩니다. 이 접미어를 빼면 클래스를 찾지 못해 커넥터 생성이 실패합니다.

name=local-file-source
connector.class=FileStreamSource
tasks.max=1
file=test.txt
topic=connect-test

# 적용 순서가 곧 이 목록의 순서입니다
transforms=MakeMap, InsertSource

transforms.MakeMap.type=org.apache.kafka.connect.transforms.HoistField$Value
transforms.MakeMap.field=line

transforms.InsertSource.type=org.apache.kafka.connect.transforms.InsertField$Value
transforms.InsertSource.static.field=data_source
transforms.InsertSource.static.value=test-file-source

변환 전
"foo"
"bar"
"hello world"

변환 후
{"line":"foo","data_source":"test-file-source"}
{"line":"bar","data_source":"test-file-source"}
{"line":"hello world","data_source":"test-file-source"}

Predicate — 조건부 SMT

SMT를 조건을 만족하는 레코드에만 적용하려면 predicate를 씁니다. 모든 SMT는 predicate와 negate라는 암묵적 설정을 가지며, negate=true로 조건을 뒤집을 수 있습니다. Apache Kafka는 세 개의 predicate를 제공합니다.

transforms=Filter,Extract

transforms.Filter.type=org.apache.kafka.connect.transforms.Filter
transforms.Filter.predicate=IsFoo

transforms.Extract.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.Extract.field=other_field
transforms.Extract.predicate=IsBar
transforms.Extract.negate=true

predicates=IsFoo,IsBar
predicates.IsFoo.type=org.apache.kafka.connect.transforms.predicates.TopicNameMatches
predicates.IsFoo.pattern=foo
predicates.IsBar.type=org.apache.kafka.connect.transforms.predicates.TopicNameMatches
predicates.IsBar.pattern=bar

에러 처리와 DLQ

공식 문서의 기본 동작 서술은 명확합니다 — 변환(conversion)이나 transform 도중 발생한 오류는 기본적으로 커넥터를 실패시킵니다("fail fast"). 이 동작은 아래 네 설정의 기본값이 만들어 냅니다.

# 실패 시 재시도하지 않음
errors.retry.timeout=0
# 오류와 그 문맥을 로그에 남기지 않음
errors.log.enable=false
# 오류 레코드를 DLQ 토픽에 기록하지 않음
errors.deadletterqueue.topic.name=
# 첫 오류에서 실패
errors.tolerance=none

# 최대 10분간 재시도, 연속 재시도 간격은 최대 30초
errors.retry.timeout=600000
errors.retry.delay.max.ms=30000

# 오류 문맥을 로그에 남기되 레코드 내용은 제외
errors.log.enable=true
errors.log.include.messages=false

# 오류 문맥을 Kafka 토픽으로 보냄
errors.deadletterqueue.topic.name=my-connector-errors

# 모든 오류를 허용 (실패시키지 않고 보고만 함)
errors.tolerance=all

DLQ 헤더로 원인을 읽는 법

errors.deadletterqueue.context.headers.enable=true일 때 붙는 헤더입니다. 원본 레코드의 헤더와 충돌하지 않도록 모두 __connect.errors. 접두어를 갖습니다.

REST API

Connect는 서비스로 실행되는 것을 전제로 하기 때문에, 커넥터 관리를 REST API로 합니다. 이 API는 standalone과 distributed 양쪽에서 모두 제공됩니다. listeners를 지정하지 않으면 기본값은 http://:8083입니다.

409 Conflict가 나는 이유

distributed 모드에서 REST API는 사용자 인터페이스이면서 워커 사이의 클러스터 내부 통신 채널이기도 합니다. 팔로워 워커가 받은 일부 요청은 리더 워커로 전달(forward)됩니다. 409 Conflict는 이 구조와 리밸런스가 만나는 지점에서 나옵니다.

CONNECT=http://connect-1:8083

# 1. 커넥터와 태스크 상태를 함께 본다. 커넥터만 보면 안 됩니다
curl -s $CONNECT/connectors/pg-source/status

# 2. FAILED 태스크의 trace 를 읽는다 (실패 원인과 어느 워커였는지)
curl -s $CONNECT/connectors/pg-source/status \
  | grep -o '"trace":"[^"]*"' | head -1

# 3. 실패한 태스크만 재시작
curl -s -X POST "$CONNECT/connectors/pg-source/restart?includeTasks=true&onlyFailed=true"

# 4. 원인 추적이 더 필요하면 런타임에 로그 레벨을 올린다 (재시작 불필요)
curl -s -X PUT -H "Content-Type: application/json" \
  --data '{"level":"DEBUG"}' \
  $CONNECT/admin/loggers/org.apache.kafka.connect.runtime.WorkerSinkTask

# 5. 오프셋을 되돌려야 하면 먼저 STOPPED 로
curl -s -X PUT $CONNECT/connectors/pg-source/stop
curl -s -X DELETE $CONNECT/connectors/pg-source/offsets
curl -s -X PUT $CONNECT/connectors/pg-source/resume

태스크 재분배와 확장 전략

공식 문서에 따르면 리밸런스는 다음 상황에 발생합니다 — 커넥터가 클러스터에 처음 제출될 때, 커넥터가 요구하는 태스크 수가 늘거나 줄 때, 커넥터 설정이 변경될 때, 그리고 워커가 그룹에 추가되거나 제거될 때 (의도적 업그레이드든 장애든).

증분 협력적 리밸런스

2.3.0 이전에는 리밸런스마다 클러스터의 모든 커넥터와 태스크를 전부 재배치했습니다. 2.3.0부터 기본값이 증분 협력적 리밸런스(incremental cooperative rebalancing)로 바뀌어, 새로 생기거나 없어지거나 이동해야 하는 태스크만 영향을 받습니다. 나머지 태스크는 멈추지도 재시작하지도 않습니다. connect.protocol의 4.3 기본값은 sessioned이며, eager로 두면 옛 전량 재배치 동작으로 돌아갑니다.

확장의 실제 상한

Connect의 병렬성은 세 단계로 제한됩니다.

1. 커넥터가 만들 수 있는 태스크 수 — tasks.max는 상한일 뿐이고, 커넥터가 소스를 그만큼 쪼갤 수 없으면 태스크가 덜 생깁니다. sink 커넥터의 실질 상한은 구독 토픽의 총 파티션 수입니다.
2. 워커 수 — 태스크는 워커에 배치되므로 워커가 적으면 한 프로세스에 태스크가 몰립니다.
3. 목적지 시스템의 수용력 — 대부분의 실제 병목은 여기입니다.

Connect에서의 exactly-once

공식 문서는 sink 커넥터의 exactly-once는 0.11.0부터, source 커넥터는 3.3.0부터 지원한다고 밝히면서, 동시에 "exactly-once 지원은 실행하는 커넥터 종류에 크게 의존한다"고 경고합니다. 워커 설정을 모두 올바르게 맞추더라도, 커넥터 자체가 프레임워크의 기능을 활용하도록 설계되지 않았다면 exactly-once가 성립하지 않습니다.

무엇을 모니터링해야 하는가

Connect는 JMX로 메트릭을 노출합니다. 장애를 조기에 잡는 데 실제로 쓰이는 것들만 골랐습니다. 전체 메트릭 목록과 알림 임계값 정리는 JMX 메트릭 치트시트에 있습니다.

보안에서 놓치기 쉬운 것

Connect 워커의 principal은 최소한 다음 ACL이 필요합니다 — Connect 클러스터 group.id에 대한 Group Read, 내부 토픽 3종에 대한 Topic Read·Write (토픽이 아직 없으면 Create까지). 커넥터별로는 source가 대상 토픽에 Write, sink가 connect-{커넥터명} 그룹에 Read와 구독 토픽에 Read, 그리고 DLQ 토픽에 Write가 필요합니다. DLQ 권한을 빼먹어 DLQ 쓰기가 실패하는 것이 흔한 초기 설정 실수입니다.

시험 포인트 정리

확인 문제

네 가지 문항 유형(단일 선택 · 복수 선택 · 연결형 · 순서 배열)이 섞여 있습니다. 오프셋 저장 위치와 409 원인은 반드시 맞히고 넘어가세요.

이어서 볼 곳

• 10장 · Kafka Streams와 ksqlDB SMT로 못 하는 집계·조인·윈도우를 어디서 처리할지. CCDAK Kafka Streams
• 예제 8 · Connect CDC 파이프라인 동작하는 source→sink 구성과 오프셋 확인 절차. 실무 예제
• 예제 6 · DLQ + 재시도 패턴 errors.* 조합을 실제로 돌려 보고 DLQ 헤더를 읽습니다. 실무 예제
• Connect 치트시트 REST 엔드포인트 전체 표, SMT 목록, 상태 코드 대응. 빠른참조
• CCDAK · Kafka Connect 도메인 15% 도메인의 압축 정리와 자주 나오는 함정. 자격증
• 트러블슈팅 치트시트 증상 → 확인 명령 → 원인 후보 결정 트리. 빠른참조

공식 문서 출처

이 장의 설정 기본값·엔드포인트·클래스명은 모두 아래에서 확인했습니다 (Apache Kafka 4.3 문서 기준).

• Kafka Connect — Overview — Connect의 목적과 특징 6가지
• Running Kafka Connect — standalone/distributed 차이, 워커 필수 설정, 내부 토픽 권장 구성, producer./consumer. 접두어 규칙
• Transformations — SMT 목록과 각 파라미터, predicate 3종, negate
• Connect REST API — 엔드포인트 전체, listeners 기본값 http://:8083, admin.listeners, 오프셋 관리 엔드포인트(KIP-875)
• Error Reporting in Connect — fail-fast 기본값 4개, DLQ 설정, errors.log.include.messages의 위험
• Exactly-once support — sink 0.11.0 / source 3.3.0, distributed 전용, preparing → enabled 2단계 롤링
• Connect Administration — 리밸런스 발생 조건, 증분 협력적 리밸런스(2.3.0), scheduled.rebalance.max.delay.ms 5분, 상태 6종, 409 Conflict, pause/stop 차이
• Connect Security — REST API 무인증 경고, connector.client.config.override.policy, 워커·커넥터별 ACL 표
• Kafka Connect Configs — offset.storage.partitions 25, status.storage.partitions 5, connect.protocol sessioned, plugin.discovery hybrid_warn, offset.flush.*
• Sink Connector Configs — errors.* 기본값, DLQ 3종이 sink 전용임
• Source Connector Configs — exactly.once.support, transaction.boundary, offsets.storage.topic
• Connect Monitoring — sink-record-lag-max, task-error-metrics, rebalancing 등 MBean
• DeadLetterQueueReporter (Apache Kafka 4.3 소스) — __connect.errors.* 헤더 키 10개