Kafka - 10. kafka connect

Kafka Connect — 코드 없이 DB·파일과 Kafka 사이 데이터를 옮기다

"DB의 주문 데이터를 Kafka로 옮기고 싶다." producer 코드를 직접 짤 수도 있지만 — 재시작 시 어디까지 읽었는지, 병렬로 어떻게 나눌지, 장애 시 복구를 다 손봐야 한다. Kafka Connect는 이 공통 작업을 프레임워크가 대신 처리한다. 커넥터 설정 하나로 데이터 파이프라인이 완성된다.

이 글은 Connect가 뭔지부터 시작해, 실제로 어떻게 구성하고 실행하는지(설정 파일, connector JSON, REST API 관리)까지 다룬다. 읽고 나면 "파일→Kafka→파일" 파이프라인을 직접 구성할 수 있다.

Connect란 — 커넥터 기반 데이터 파이프라인 프레임워크

Connect는 데이터를 Kafka와 외부 시스템(DB·파일·S3) 사이로 옮기는 프레임워크다. 핵심 개념:

• connector — "어디서→어디로, 무엇을"의 정의. 두 종류:
  • source connector: 외부 시스템 → Kafka (DB 변경을 topic으로).
  • sink connector: Kafka → 외부 시스템 (topic 데이터를 ES·S3로).
• task — connector가 작업을 병렬 단위로 쪼갠 것. worker들이 나눠 실행.
• worker — Connect 프로세스.

flowchart LR
    DB[(Postgres DB)] -->|source connector| K[Kafka topic]
    K -->|sink connector| ES[(Elasticsearch)]
    K -->|sink connector| S3[(S3)]
    subgraph CW[Connect worker]
        C[connector] -->|작업 분할| T1[task 1]
        C --> T2[task 2]
    end

Connect는 Apache Kafka core 배포판에 포함된다(bin/connect-*). 별도 설치 불필요. 단, Schema Registry는 Confluent 확장이라 별도(장 09).

standalone vs distributed — 어느 모드를 쓸까

모드	용도	상태 저장	특징
standalone	개발/테스트	로컬 파일	단일 프로세스. 단일 장애점.
distributed	프로덕션	내부 topic	여러 worker가 task 공유. 장애 시 자동 재배치. REST API 관리.

프로덕션은 무조건 distributed — standalone은 단일 장애점이라 장애 시 데이터 이동이 멈춘다.

Connect 구성 방법 — 설정 파일부터 시작

Connect를 실행하려면 worker 설정 + connector 설정 두 가지가 필요하다.

worker 설정 (standalone)

config/connect-standalone.properties — Connect 프로세스 자체의 설정:

# Kafka 클러스터 연결
bootstrap.servers=localhost:9092

# 데이터 직렬화 포맷 (converter)
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false

# offset/config/status 저장 위치 (standalone은 파일)
offset.storage.file.filename=/tmp/connect.offsets

# REST API 포트 (distributed에서 필수, standalone은 선택)
rest.port=8083

# 커넥터 플러그인 경로 (JAR 배치 위치)
plugin.path=/usr/local/share/kafka/plugins

key.converter/value.converter가 핵심 — Connect가 Kafka에 쓸 때 바이트로 변환하는 포맷. JSON 또는 Avro(+Schema Registry). 잘못된 converter는 consumer가 못 읽는 바이트를 만든다.

worker 설정 (distributed)

config/connect-distributed.properties — 차이점은 상태를 Kafka 내부 topic에 저장:

bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

# 상태 저장용 내부 topic (distributed 모드 필수)
group.id=connect-cluster
offset.storage.topic=connect-offsets
config.storage.topic=connect-configs
status.storage.topic=connect-status

# 내부 topic 설정 (compact 권장 - 최신 설정만 유지)
offset.storage.replication.factor=3
config.storage.replication.factor=3
status.storage.replication.factor=3

rest.port=8083
plugin.path=/usr/local/share/kafka/plugins

connect-offsets/connect-configs/connect-status는 삭제·retention 변경 금지. Connect가 상태를 의존한다.

distributed 모드 기동

# Kafka 4.3, KRaft 단일 노드 기동 후
bin/connect-distributed.sh config/connect-distributed.properties
# 여러 터미널에서 같은 설정으로 추가 기동 → 자동 클러스터 형성

connector 구성 — 어떻게 데이터 파이프라인을 정의하나

방식 1: properties 파일 (standalone)

config/connect-file-source.properties:

name=file-source
connector.class=org.apache.kafka.connect.file.FileStreamSourceConnector
tasks.max=1
file=/tmp/test.txt
topic=connect-test

방식 2: REST API + JSON (distributed, 권장)

distributed 모드에선 REST API(포트 8083)로 connector를 관리한다:

connector 생성:

curl -X POST -H "Content-Type: application/json" \
  --data '{
    "name": "file-source-demo",
    "config": {
      "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
      "tasks.max": "1",
      "file": "/tmp/test.txt",
      "topic": "connect-test"
    }
  }' \
  http://localhost:8083/connectors

JDBC source connector 예 (DB → Kafka):

curl -X POST -H "Content-Type: application/json" \
  --data '{
    "name": "postgres-orders-source",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
      "connection.url": "jdbc:postgresql://localhost:5432/mydb",
      "connection.user": "postgres",
      "connection.password": "secret",
      "table.whitelist": "orders",
      "mode": "incrementing",
      "incrementing.column.name": "id",
      "topic.prefix": "postgres-",
      "tasks.max": "1"
    }
  }' \
  http://localhost:8083/connectors

이 설정으로 Postgres orders 테이블의 새 행이 postgres-orders topic으로 자동 들어간다.

REST API — connector 생명주기 관리

distributed 모드의 REST API(포트 8083)로 모든 관리를 한다:

작업	명령
목록	GET /connectors
상세	GET /connectors/{name}
상태	GET /connectors/{name}/status
생성	POST /connectors (JSON body)
설정 변경	PUT /connectors/{name}/config (JSON body)
일시정지	PUT /connectors/{name}/pause
재개	PUT /connectors/{name}/resume
재시작	POST /connectors/{name}/restart
task 재시작	POST /connectors/{name}/tasks/{taskId}/restart
삭제	DELETE /connectors/{name}
플러그인 목록	GET /connector-plugins

상태 확인 예

curl http://localhost:8083/connectors/file-source-demo/status

{"name":"file-source-demo","connector":{"state":"RUNNING","worker_id":"localhost:8083"},
 "tasks":[{"state":"RUNNING","worker_id":"localhost:8083"}]}

확인할 것: RUNNING/FAILED/PAUSED. FAILED 시 trace 필드로 원인 확인.

일시정지·재개

# 일시정지 (데이터 이동 중단, connector는 유지)
curl -X PUT http://localhost:8083/connectors/file-source-demo/pause
# 재개
curl -X PUT http://localhost:8083/connectors/file-source-demo/resume

플러그인 확인

curl http://localhost:8083/connector-plugins | jq '.[].class'

plugin.path에 배치한 커넥터 JAR들이 보여야 한다.

커넥터 설치 — 플러그인 관리

Connect는 plugin.path에 있는 JAR을 자동으로 인식한다:

# 커넥터 JAR을 plugin.path에 배치
cp my-connector.jar /usr/local/share/kafka/plugins/
# Connect 재기동 (또는 분산 모드면 자동 감지)

Confluent Hub로 커넥터 설치:

confluent-hub install confluentinc/kafka-connect-jdbc:latest
# → plugin.path에 자동 설치

대표 커넥터:

• Debezium (CDC: MySQL/Postgres 변경 감지)
• JDBC Source/Sink (DB ↔ Kafka)
• Elasticsearch Sink (Kafka → ES)
• S3 Sink (Kafka → S3)
• FileStream (파일 ↔ Kafka, 학습용)

converter와 SMT — 메시지 변환

converter

Connect는 내부적으로 SourceRecord/SinkRecord 모델을 쓴다. Kafka에 쓸 땐 converter가 바이트로 변환:

• JsonConverter — JSON 직렬화 (디버깅 쉬움).
• AvroConverter — Avro + Schema Registry 연동 (장 09).
  잘못된 converter는 consumer가 못 읽는 바이트를 만든다 — "Unknown magic byte!" 에러의 원인.

SMT (Single Message Transform)

메시지 하나씩 변환(필드명 변경·타입 변환·timestamp 추가):

"transforms": "RenameField",
"transforms.RenameField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.RenameField.renames": "[\"old_name:new_name\"]"

복잡 집계·조인은 불가 → 그건 Kafka Streams(장 11).

실습 — FileStream 파이프라인 구성하기

standalone으로 파일→Kafka→파일

# 1. 입력 파일 준비
echo "hello connect" > /tmp/test.txt

# 2. Connect 기동 (file-source + file-sink 동시)
bin/connect-standalone.sh config/connect-standalone.properties \
  config/connect-file-source.properties \
  config/connect-file-sink.properties

# 3. test.txt에 줄 추가 → connect-test topic → 출력 파일에 나타남
echo "second line" >> /tmp/test.txt
cat /tmp/test.sink.txt  # 출력 확인

distributed로 REST API로 connector 관리

# 1. Connect distributed 기동
bin/connect-distributed.sh config/connect-distributed.properties

# 2. REST로 connector 생성 (위 JSON 예 참조)

# 3. 상태 확인
curl http://localhost:8083/connectors

# 4. 파일에 데이터 추가하고 topic/출력 파일 확인

흔히 묻는 것, 흔히 틀리는 것

오해	정정
"Connect는 Kafka core가 아니다"	core 포함. Schema Registry만 별도
"standalone으로 프로덕션 해도 된다"	단일 장애점. 프로덕션은 distributed
"converter는 아무거나 해도 된다"	source/consumer 호환 포맷이어야
"SMT로 복잡 집계를 한다"	SMT는 메시지 단위 경량 변환만. 집계는 Streams
"task 수는 많을수록 좋다"	source/sink 대역폭·파티션 수에 bound
"내부 topic은 삭제해도 된다"	connect-offsets/configs/status 삭제 금지

더 깊이

• DLQ (Dead Letter Queue): 처리 실패 메시지를 별도 topic(errors.deadletterqueue.topic.name)으로 보내 분석.
• exactly-once Connect: source 커넥터는 at-least-once가 기본. sink는 멱등 쓰기 권장.
• 커넥터 개발: Connector + Task 인터페이스 구현으로 커스텀 커넥터 작성 가능.

요약 — 이 글의 결론

• Kafka Connect = 코드 없는 데이터 이동 프레임워크(Kafka core 포함).
• source(외부→Kafka) / sink(Kafka→외부). connector가 작업을 task로 분할.
• worker 설정: standalone(개발, 파일에 상태) / distributed(프로덕션, 내부 topic에 상태, REST 8083).
• connector 구성: properties 파일(standalone) 또는 REST API + JSON(distributed).
• REST API CRUD: 생성·상태·일시정지·재개·재시작·삭제.
• converter: wire format 결정. SMT는 메시지 단위 변환.
• 플러그인: plugin.path에 JAR 배치 또는 Confluent Hub 설치.
• Connect(데이터 이동) ≠ Streams(데이터 가공).

생각해 볼 문제

1. source 커넥터와 sink 커넥터의 차이를 설명하라.
2. standalone과 distributed의 차이. 프로덕션은 왜 distributed인가?
3. REST API로 connector를 생성하는 JSON을 직접 작성해 보라.
4. converter를 잘못 설정하면 어떤 일이 벌어지는가?
5. SMT가 할 수 있는 것과 할 수 없는 것은? 안 되는 건 무엇으로 푸는가?
6. distributed 모드의 내부 topic(connect-offsets 등)이 왜 중요한가?

---

참고

• Kafka Connect — User Guide - 접근 2026-07-09
• Connector Development Guide - 접근 2026-07-09
• Quickstart — FileStream 파이프라인 - 접근 2026-07-09