Kafka Connect の概要: データ連携の標準基盤

Kafka Connect は、外部システムと Kafka を低コードでつなぐプラガブルなフレームワーク。Source/Sink コネクタとタスクをワーカーにデプロイしてスケールさせるだけで、信頼性のあるデータ連携基盤を構築できる。

本稿は CCDAK の出題観点（モード選択、内部トピック、タスク/コネクタ、コンバータ/SMT、エラーハンドリング）と、実務の運用手順を両立して解説する。

Kafka Connect の全体像と位置づけ

Kafka Connect は、外部システム（DB、オブジェクトストレージ、SaaS など）とのデータ連携を、コネクタというプラグインとして実装・運用するための公式フレームワーク。アプリケーションに専用の Producer/Consumer を書かずに、設定ベースで連携を行う。

ワーカー（Connect クラスタ）はタスクを並列実行し、分散モードでは構成・オフセット・ステータスを Kafka の内部トピックで耐障害的に管理する。操作は REST API 経由で行い、ローリングでの拡張・更新が可能。

• Source コネクタ: 外部→Kafka へ取り込み
• Sink コネクタ: Kafka→外部へ配信
• タスク: コネクタの並列実行単位（スループットとフォールトトレランスの鍵）
• コンバータと SMT: シリアライズ/変換の責務を分離して再利用性を高める

Kafka Connect の論理アーキテクチャ

最小の Source コネクタを作成（REST）

curl -s -X POST http://localhost:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "fs-source",
    "config": {
      "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
      "tasks.max": "1",
      "file": "/tmp/input.txt",
      "topic": "fs-input",
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
      "value.converter": "org.apache.kafka.connect.storage.StringConverter"
    }
  }'

実行モードの比較と選択基準

Kafka Connect は Standalone と Distributed の2モードを持つ。前者は単一プロセスで簡易、後者はクラスタリングして耐障害・スケールアウトを実現する。マネージド（例: Confluent Cloud）の提供もあり、運用をアウトソースできる。

分散モードでは内部トピック（config/offsets/status）により状態が Kafka 上に保持され、ワーカーの増減や障害時も自動でリバランスされる。スタンドアロンは開発・単体用途に向く。

• 本番は原則 Distributed。ローカル開発や PoC は Standalone で素早く開始
• レイテンシ要件が厳しい場合も、まずはタスク並列化と十分なパーティション設計で解決を検討
• マネージドは SLA/運用体制/接続要件（VPC Peering 等）を含め総合判断

起動コマンドの対比（例）

# Standalone（単一プロセス）
connect-standalone worker.properties file-source.properties

# Distributed（ワーカークラスタ）
connect-distributed worker.properties  # その後 REST でコネクタ作成

コンポーネント: コネクタ / タスク / コンバータ / SMT

コネクタは「外部システムとの接続と分割統治」の責務、タスクは「並列実行単位」の責務を持つ。tasks.max に応じてタスクが生成され、分散モードでは複数ワーカーに割り当てられる。

コンバータ（key/value.converter）は Connect と Kafka 間のシリアライズ形式を定義する。JSON/String/ByteArray が同梱され、Avro/Protobuf などは一般に別途プラグインとスキーマ管理（例: Schema Registry）を用いる。SMT は単一レコード単位の軽量変換チェーンで、配線前後のデータ整形に有効。

• タスクはスループットと可用性の基本単位。スケール時は tasks.max と入力パーティション数の関係を意識
• SMT は軽量変換。複雑な処理は Kafka Streams や下流で実装
• コンバータは Connect 内部表現と Kafka バイト配列の橋渡し。スキーマ有無の設定に注意

SMT とコンバータの設定例（抜粋）

{
  "name": "jdbc-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topics": "orders",
    "tasks.max": "4",
    "connection.url": "jdbc:postgresql://db:5432/app",
    "insert.mode": "upsert",
    "pk.mode": "record_key",

    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",

    "transforms": "route,mask",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "(.*)",
    "transforms.route.replacement": "prod_$1",
    "transforms.mask.type": "org.apache.kafka.connect.transforms.MaskField$Value",
    "transforms.mask.fields": "card_number"
  }
}

内部トピックと信頼性の仕組み

分散モードでは、connect-configs・connect-offsets・connect-status の内部トピックが用意される。いずれもログコンパクション対象で、推奨は十分なレプリケーション係数（一般に 3）を設定して耐障害性を確保すること。

Source のオフセットは外部システム側の位置情報（ソースパーティション＋位置）をマップとして保持し、Sink は Kafka Consumer のオフセットを保持する。これにより、ワーカー障害・再起動後も継続位置から再開できる。

トピック名はワーカー構成で指定し、事前作成しない場合は自動作成されることもあるが、本番ではポリシーどおりに手動作成し、コンパクションとレプリケーションを明示するのが安全。

• config.storage.topic / offsets.storage.topic / status.storage.topic を明示
• 各 internal topic に適切な replication.factor を設定
• ワーカー増減やローリング再起動時のリバランスに備え、安定したブローカー側設定と監視を用意

分散ワーカー設定（抜粋）

# Kafka 接続
bootstrap.servers=broker1:9092,broker2:9092

group.id=connect-cluster

# 内部トピック
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
config.storage.replication.factor=3
offset.storage.replication.factor=3
status.storage.replication.factor=3

# コンバータ（例: スキーマなし JSON）
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

# REST ポート
rest.port=8083

# （必要に応じて）セキュリティや Config Provider
# ssl.endpoint.identification.algorithm=https
# sasl.mechanism=PLAIN
# config.providers=env,file
# config.providers.env.class=org.apache.kafka.common.config.provider.EnvVarConfigProvider

運用: スケーリング、監視、エラーハンドリング

スケールは tasks.max の調整とワーカー台数の増減で行う。リバランスによりタスク割当が自動再配置される。入力トピックのパーティション数以上に Sink タスクを増やしても効果は薄いため、パーティション設計と併せ込む。

監視は REST と JMX を併用。/connectors、/connectors/{name}/status、/connectors/{name}/tasks などで稼働状態を確認できる。失敗時の挙動は errors.* 設定で制御でき、DLQ（デッドレタートピック）への退避も可能。

ローリング更新は 1 台ずつワーカーを再起動し、各再起動ごとにステータスを確認する。長時間のリバランスやタスク起動失敗が続く場合は、内部トピックの可用性と外部依存（DB 接続、認証）の健全性を優先確認する。

• Sink: tasks.max は入力パーティション数を上限目安に設定
• Source: 外部側シャーディングと tasks.max の整合を取る
• エラー処理は retry + DLQ の二段構え。恒久的失敗は DLQ で隔離し後処理
• REST による Pause/Resume を運用 Runbook に組み込む

DLQ と監視 REST（例）

# 監視
curl -s http://localhost:8083/connectors | jq .
curl -s http://localhost:8083/connectors/my-conn/status | jq .
curl -s http://localhost:8083/connectors/my-conn/tasks | jq .

# 一時停止/再開
curl -s -X PUT http://localhost:8083/connectors/my-conn/pause
curl -s -X PUT http://localhost:8083/connectors/my-conn/resume

# エラー処理（コネクタ設定の一部）
"errors.tolerance": "all",                      # all または none
"errors.retry.timeout": "60000",                # ms
"errors.retry.delay.max.ms": "5000",
"errors.deadletterqueue.topic.name": "dlq.my-conn",
"errors.deadletterqueue.context.headers.enable": "true"

CCDAK 向け要点と設計パターン

出題では、モード選択、内部トピックの役割、タスクとパーティションの関係、コンバータ/SMT の責務分離、エラーハンドリングの構成が頻出。設計問題では、少なくとも一度の配送前提で冪等な下流更新（Upsert/Idempotent API）を選ぶ、といった方針が問われやすい。

一般的なパターンは CDC Source → Kafka → 複数 Sink（DWH、検索、レイク）。スキーマ進化を考慮し、スキーマ管理と互換性戦略（後方互換など）を決める。性能面は、タスク並列化・適切なパーティション数・バッチサイズやフラッシュ間隔の調整で詰める。

• タスク ≤ 入力パーティション数（Sink の基本）。Source は外部シャーディングに合わせる
• 内部トピックはコンパクション＋十分なレプリケーション
• Connect は原則 at-least-once。整合性は外部側の冪等化や再処理設計で担保
• SMT は軽量変換。重い加工は別レイヤー（Kafka Streams など）に逃がす
• セキュリティとシークレットはワーカー設定で一元管理

スケーリング設計のメモ（概算）

# 目標スループットとパーティション数から tasks.max を見積もる
# P = 入力パーティション数, Rt = 目標総処理行/秒, Rc = 1タスクあたり処理行/秒
# tasks.max ≈ min(P, ceil(Rt / Rc))
# 例: P=24, Rt=48k rps, Rc=3k rps → tasks.max=min(24, ceil(48000/3000)=16)=16

問題で確認

よくある質問