Kafka Connect の分散モードとスタンドアロン: 構成選択と運用の勘所

Kafka Connect はコネクタの実行基盤として、スタンドアロンと分散モードの2形態を提供します。どちらも同じコネクタ/タスクのモデルを使いますが、構成の保管先やフェイルオーバー、スケール方法が異なります。

本稿は、選定の指針と運用ベストプラクティスを平易に整理し、試験（CCDAK/CCAAK）で頻出のポイントもあわせて確認します。公式ドキュメントに基づく安定した概念に絞って解説します。

分散モードとスタンドアロンの基本と選定

スタンドアロンは1プロセス内でコネクタとタスクを実行し、設定・オフセットをローカルファイルに保持します。フェイルオーバーやリバランスはありません。単発の移行、開発検証、1台限定のバッチ処理などに向きます。

分散モードは複数の Connect ワーカーで1つのクラスタを組み、構成・オフセット・ステータスをKafkaの内部トピックに保持します。ワーカー障害時はタスクが他ワーカーへ再割当てされ、スケールアウトやローリングアップグレードが可能です。本番運用の既定選択はこちらです。

• 可用性が必要なら分散モードを選ぶ
• 単発/小規模で速く始めたいならスタンドアロン
• 将来のスケールを見込むなら最初から分散モードに寄せる

Kafka Connect: 分散モードとスタンドアロンの俯瞰

最小構成の対比（スタンドアロン vs 分散ワーカー）

# standalone.properties（抜粋）
bootstrap.servers=broker1:9092
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
offset.storage.file.filename=/var/lib/kafka-connect/connect.offsets
offset.flush.interval.ms=10000

# worker.properties（分散モード, 抜粋）
bootstrap.servers=broker1:9092,broker2:9092
group.id=connect-cluster-1
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

ワーカー、コネクタ、タスク、内部トピックの役割

Connect では、ワーカーがプロセス、コネクタがジョブ、タスクが並列実行単位です。タスク数はコネクタ設定（tasks.max）やシンクならパーティション数により上限が決まります。

分散モードでは、構成（config.storage.topic）、オフセット（offset.storage.topic）、ステータス（status.storage.topic）をKafka内部トピックに保存します。これにより、REST経由で投入した設定がクラスタ全体で共有され、障害復旧やローリングアップグレードが容易になります。内部トピックはコンパクション有効で作成され、レプリケーションファクターは本番で少なくとも3を推奨します。自動作成を禁じているクラスタでは、事前に適切な設定で作成しておきます。

• config.storage.topic: コネクタ設定の単一ソース
• offset.storage.topic: ソース/シンクの進捗を保持（コンパクション）
• status.storage.topic: 実行状態監視に利用。管理系ツールが参照
• 事前作成時の推奨: cleanup.policy=compact、RF>=3、パーティション数はワークロードに応じ決定（例: config=1, status=5, offsets=25 など）

内部トピック関連の代表的プロパティ（分散ワーカー）

bootstrap.servers=broker1:9092,broker2:9092
group.id=connect-cluster-1
config.storage.topic=connect-configs
config.storage.replication.factor=3
offset.storage.topic=connect-offsets
offset.storage.replication.factor=3
status.storage.topic=connect-status
status.storage.replication.factor=3

フォールトトレランスと復旧手順

分散モードでは、ワーカー障害時にそのワーカーが保持していたタスクがクラスタ内の健全なワーカーへ再割当てされます。コネクタ設定は内部トピックにあるため、プロセスが変わっても継続実行できます。スタンドアロンは単一プロセスのため、障害でタスクが停止し、手動復旧が必要です。

設定変更は REST API で投入し、分散クラスタの全ワーカーが共有します。復旧時はコネクタやタスクの再起動、必要に応じて一時停止/再開を使います。

• 分散モード: ワーカー追加/削除で自動リバランス
• スタンドアロン: 再起動とローカルオフセットファイルの健全性確認が必須
• 変更は REST から。ローカル編集/再起動が必要なのは主にスタンドアロン

Connect REST API による運用（例）

# コネクタ一覧
curl -s http://connect1:8083/connectors

# コネクタ作成
curl -s -X POST http://connect1:8083/connectors -H 'Content-Type: application/json' -d '{
  "name": "jdbc-sink-01",
  "config": {"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "topics": "orders", "tasks.max": "4"}
}'

# 一時停止/再開
curl -X PUT http://connect1:8083/connectors/jdbc-sink-01/pause
curl -X PUT http://connect1:8083/connectors/jdbc-sink-01/resume

# 再起動（コネクタ/すべてのタスク）
curl -X POST http://connect1:8083/connectors/jdbc-sink-01/restart?includeTasks=true&onlyFailed=true

スケーリングとスループット設計

シンクの並列度は原則として対象トピックのパーティション数に制限され、tasks.max をそれ以下に設定します。ソースは接続先システムの分割性に依存します。分散モードではワーカーを追加するだけで総処理能力を伸ばせます。

スタンドアロンでスケールしたい場合は、別プロセスとして複数起動し、対象トピックやテーブルを明確に分割する必要があります（オフセットファイルも分離）。一方、分散モードでは1つの設定を REST で投入し、クラスタが自動でタスク配分します。

• シンク: tasks.max ≤ 総パーティション数
• ソース: 入力側のシャーディング単位に合わせて tasks.max を設計
• ワーカー台数の増減で処理能力を調整（分散モード）

シンク・コネクタ設定例（スループット調整）

{
  "name": "s3-sink-raw",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "topics": "raw.events",
    "tasks.max": "6",
    "flush.size": "10000",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}

運用: 監視・ログ・セキュリティ・アップグレード

監視は JMX メトリクス（ワーカー/コネクタ/タスク単位）とステータストピックの併用が実務的です。エラーハンドリングは Dead Letter Queue（DLQ）やログ出力を適切に構成します。

セキュリティはブローカーへの接続（SASL/SSL）をワーカー共通のプロパティ、あるいはコネクタごとの producer.* / consumer.* で上書き設定します。分散モードのアップグレードはローリングが基本で、ワーカーを1台ずつ停止・更新・復帰させます。

• JMX で connect-worker-metrics, connect-task-metrics を収集
• DLQ（errors.deadletterqueue.topic.name）とエラーログ（errors.log.enable）を併用
• ローリングアップグレードではコネクタの pause/resume も活用

代表的な運用設定スニペット

# JMX 有効化（起動スクリプトなど）
export KAFKA_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=9010 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false"

# DLQ とエラー制御（コネクタ側設定）
errors.tolerance=all
errors.log.enable=true
errors.deadletterqueue.topic.name=connect-dlq
errors.deadletterqueue.context.headers.enable=true

# ブローカー接続のセキュリティ例（ワーカー共通）
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="user" password="pass";

試験対策の要点チェック（CCDAK / CCAAK）

モード差分と内部トピックの役割、オフセットの保存先、スケーリングの制約（tasks.max とパーティション数の関係）は頻出です。REST による運用、DLQ とエラー制御、ローリングアップグレードの可否も問われがちです。

用語の区別（ワーカー / コネクタ / タスク）と、分散モードでの設定共有の仕組み（config.storage.topic）が正しく説明できることが合格点の目安です。

• 分散モードのみが内部トピック（config/offset/status）を使用する
• スタンドアロンのオフセットはローカルファイル。可用性はワーカー単体に依存
• シンクの並列度はパーティション数以下、ソースは入力側の分割性依存
• REST API で構成投入・更新・再起動・一時停止/再開が可能
• 本番はレプリケーションファクター>=3、内部トピックはコンパクション

覚えておきたい主要プロパティ早見

# 分散モード必須
group.id=connect-cluster-1
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status

# スタンドアロン固有
offset.storage.file.filename=/var/lib/kafka-connect/connect.offsets

# 並列度とフラッシュ
tasks.max=4
offset.flush.interval.ms=10000

問題で確認

よくある質問