Kafka Connectとは？構成要素とコネクタ運用を実装視点で解説【2026年版】

Kafka Connectは、Apache Kafkaに同梱されるデータ連携フレームワークです。データベースやオブジェクトストレージとKafkaの間でレコードを流す処理を、JSONの設定を投げるだけで動かせます。2026年8月時点の最新安定版は2026年6月25日公開の4.3.1で、Connectワーカーの実行にはJava 17以上が必要です。この記事では、ワーカーとコネクタとタスクの関係、分散モードの設計、REST APIでの登録と停止、DLQやexactly-onceの設定、そして採用を見送るべき条件を扱います。ブローカー側の仕組みはApache Kafkaの特徴を整理した記事に譲りました。

まとめ｜Kafka Connect採用を分ける3条件と配置モードの選び方

Kafka Connectが効いてくるのは、次の3つが揃った場合です。すでにKafkaクラスタが業務データの通り道になっていること。連携先が3系統以上あり、個別スクリプトでは保守が破綻すること。障害時にどのレコードまで流れたかをオフセットで説明する必要があること。3つ目が不要ならcronのバッチで足ります。

実行モードの選択で迷う場面は少なく、本番は分散モードの一択です。スタンドアロンは単一プロセスでフォールトトレランスを持たず、ソースコネクタのexactly-onceも分散モード限定という制約があります。分散モードでは設定・オフセット・ステータスの3トピックがKafka側に置かれ、ワーカーを足すだけでタスクが再配置されます。

逆に、連携先が1系統で日次バッチの遅延が許される案件では、ワーカーの常時稼働とJVMの監視という負荷が見合いません。採用ラインの引き方は第8章で条件付きに示しました。

ワーカー・コネクタ・タスク・コンバータで構成される4層の実行モデル

設定プロパティがどの層に属するかを取り違えると、動かない設定を延々と直すことになります。登場人物を4つに切り分けます。

ワーカープロセスが担うプラグイン読み込みとオフセットコミット

ワーカーはコネクタとタスクを走らせるJVMプロセスです。起動時にplugin.path配下を走査してコネクタ・コンバータ・変換のクラスを読み込みます。この既定値はnullで、指定しない限り同梱のFileStreamSourceConnectorすら見つかりません。

読み込み方式はplugin.discoveryで切り替わり、3.6以降の既定はhybrid_warnです。ServiceLoader非対応のプラグインがあれば起動時に警告が出るので、警告ゼロを確認のうえservice_loadに変えると起動を短縮できます。オフセットのコミット間隔はoffset.flush.interval.msで、既定は60000ミリ秒です。

コネクタとタスクの分離が生む並列度とオフセット管理の担当範囲

コネクタは「どこからどこへ運ぶか」を決める論理的なジョブで、実際にレコードを運ぶのはタスクです。コネクタが設定を読んで分割数を決め、各タスクへ担当範囲を配ります。1テーブル1タスクか、パーティション単位かはコネクタ実装しだいです。

この分離のおかげで、開発者はオフセットのコミット処理を書かずに済みます。ソースタスクは読み位置を任意のキーとバリューで返すだけでよく、永続化はフレームワーク側の仕事です。tasks.maxは上限の宣言にすぎず、コネクタが並列度を出せなければ指定より少ないタスクしか作られません。

ソースコネクタとシンクコネクタで異なる入力指定と設定必須項目

設定上の差がはっきり出るのは入力の指定方法です。シンクコネクタはtopicsかtopics.regexのどちらかを必ず設定します。ソースコネクタは送り先トピック名を固有のプロパティで持ちます。

ソースコネクタの代表例が、データベースのトランザクションログを読むDebeziumのコネクタ群です。binlogや論理レプリケーションの有効化はデータベース側の作業で、Kafka Connectの守備範囲の外にあります。前提となる権限とログ保持の設計はCDCの3方式と設定をまとめた記事で確認してください。

スタンドアロンと分散モードの違いと内部トピック3本の設計基準

2つの実行モードの違いは、状態をどこに置くかだけです。ローカルファイルに置くのがスタンドアロン、Kafkaのトピックに置くのが分散モードになります。

開発検証で使うスタンドアロンモードの起動手順と単一障害点の制約

スタンドアロンは、connect-standalone.shにワーカー設定とコネクタ設定を並べて渡すだけで起動する方式です。コネクタ設定はpropertiesでもJSONでもよく、複数指定すれば同一プロセスの別スレッドで動きます。ソースのオフセットはoffset.storage.file.filenameが指すファイルに書かれるため、そのファイルを失えば読み位置も消えます。

向くのは1台のサーバのログ収集のように、ワーカーが1つで完結する用途だけです。検証環境で設定を詰め、同じ設定を分散モードへREST経由で投入する進め方だと手戻りが出にくくなります。

分散モードで必須になるgroup.idと3つの内部トピックの推奨設定

分散モードでは共通設定のbootstrap.serversと2つのコンバータに加え、クラスタを識別するgroup.idと3本の内部トピックを指定します。group.idはコンシューマグループのIDと衝突させてはならず、命名規則を分けておくと事故を防げます。

設定用トピックが1パーティション固定なのは、全ワーカーが同じ順序で設定変更を読む必要があるからです。クラスタ自体をマネージドへ寄せるなら、Amazon MSKの仕組みと料金を整理した記事で管理責任の線引きを確認しておいてください。

内部トピックを自動作成に任せた場合に起きるパーティション不足

3本のトピックは存在しなければ起動時に自動作成されますが、公式ドキュメントは手動作成を推奨しています。自動作成ではブローカーの既定値が使われ、パーティション数もレプリケーション係数も用途に合わない値になるためです。

実害が出やすいのはオフセット用トピックです。既定の少ないパーティション数で作られると、ソースコネクタが増えたときに書き込みが特定のブローカーへ集中します。後から増やしてもキーの配置が変わって読み位置を見失う恐れがあり、作り直しが要ります。トピック3本の作成は構築手順書の先頭に置いてください。

コネクタ登録からステータス確認までのREST API運用手順

分散モードではコマンドライン引数でコネクタを渡せず、運用の入口はすべてREST APIです。listeners未指定ならHTTPのポート8083で待ち受けます。

設定JSONに書くconnector.classとtasks.maxの指定パターン

登録リクエストのボディは、文字列のnameとオブジェクトのconfigを持つJSONです。configの先頭に書くconnector.classは完全修飾名でもエイリアスでも通り、FileStreamSinkConnectorならFileStreamSinkという短縮形が使えます。

見落とされがちなのが初期状態の指定です。ボディにinitial_stateを足してSTOPPEDやPAUSEDを渡せば、設定だけ先に投入し、流し始めるタイミングを別に決められます。

主要エンドポイント一覧と停止・一時停止・オフセット操作の違い

下の表はパスから先頭のconnectorsを省いた形です。実際のURLはホストとポートの後にconnectorsを挟みます。

pauseとstopは似て非なる操作です。pauseはタスクを生かしたまま処理だけ止めるので再開が速く、3.5.0で追加されたstopはタスクを落としてリソースを解放します。そしてオフセットの変更と削除はstopped状態でなければ受け付けられません。特定の日時から流し直す要求には、stopしてPATCHでオフセットを書き換え、resumeで戻す3手順で応えます。

ステータスAPIが返す6状態と409応答が出る場面の切り分け

状態確認のレスポンスには、コネクタ本体と各タスクの状態、割り当てられたワーカーのアドレスが並びます。取りうる状態はUNASSIGNED・RUNNING・PAUSED・STOPPED・FAILED・RESTARTINGの6つで、STOPPEDはコネクタにしか現れません。停止時にタスクは消えるからです。

監視で押さえるべきは、コネクタがRUNNINGでもタスクがFAILEDになりうる点です。Connectは失敗タスクを自動再起動しないため、放置すればコネクタは正常に見えたままデータだけが止まります。アラート対象はタスク単位の状態にしてください。リバランス中の再起動要求には409が返ります。

コンバータ選択とSMTによるスキーマ変換・ルーティングの設定例

コンバータとSMTは、コネクタを書き換えずにデータの形を変えられる層です。ここを押さえると既製コネクタの射程が広がります。

JSONとAvroでスキーマ管理の負荷が変わるコンバータ設定の指針

コンバータはConnect内部のレコード表現とKafkaに書かれるバイト列を相互変換します。コネクタから独立しているため、どのコネクタとどのシリアライズ形式でも組み合わせられる設計です。ワーカー設定のkey.converterとvalue.converterが既定値になり、コネクタ側で上書きもできます。

JSONコンバータのvalue.converter.schemas.enableをtrueにすると、ペイロードにスキーマ定義が毎回同梱されます。数十フィールドのテーブルでは、実データより型定義の方が長いレコードが延々と流れかねません。長期に流し続けるならAvroやProtobufとスキーマレジストリの組み合わせに寄せ、検証段階だけschemas.enableをfalseにしたJSONで進める切り分けが扱いやすくなります。

標準SMT16種のうち実務での使用頻度が高い4種と設定の書き方

Single Message Transformsはレコード1件ごとの軽い加工を担います。4.3系には16種が同梱され、transformsに別名をカンマ区切りで並べた順に適用されます。出番が多いのは次の4つです。

• RegexRouter：正規表現で送り先トピック名を書き換える。テーブル名の接頭辞を落とす用途で頻出
• InsertField：静的な値やレコードのメタデータをフィールドとして足す。取り込み元の識別に使う
• MaskField：フィールドを型の既定値か指定値で置き換える。個人情報を落として下流へ流す場合に使う
• ValueToKey：バリューの一部フィールドから新しいキーを組み立てる。パーティション分散の調整に効く

別名は自由に付けられ、transforms.別名.typeに完全修飾クラス名を書く設定です。キー側とバリュー側で別クラスになる変換が多く、末尾の記号を取り違えると起動時に落ちます。複数レコードの結合や時間窓の集計はSMTの守備範囲を超えます。

エラー許容とDLQ・exactly-once設定で決まる配送保証の水準

既定のKafka Connectは、変換処理で1件でも失敗すればタスクを落とします。この挙動を知らずに本番へ出すと、不正な1レコードで連携が全停止します。

errors.toleranceとDLQ設定で失敗レコードを退避する構成

既定値はerrors.toleranceがnone、errors.retry.timeoutが0、errors.log.enableがfalseというfail fastです。許容側へ倒すにはerrors.toleranceをallにし、退避先としてerrors.deadletterqueue.topic.nameにトピック名を設定します。errors.deadletterqueue.context.headers.enableをtrueにすれば元のトピック・パーティション・オフセットがヘッダに残り、後追い調査が現実的な作業量に収まります。

リトライはerrors.retry.timeoutとerrors.retry.delay.max.msを組みます。公式の例は600000ミリ秒（10分）まで、間隔は最大30000ミリ秒という指定です。DLQは作って終わりではなく、滞留件数の監視と再投入の手順まで設計に含めてください。

ソースコネクタのexactly-onceを有効化する2段階ローリング更新

対応時期はシンクとソースで大きく違い、シンクは0.11.0から、ソースは3.3.0からです。シンク側はワーカー設定のconsumer.isolation.levelをread_committedにするか、上書きポリシー経由でコネクタ単位に同じ値を渡します。

ソース側は手順が重くなります。分散モード限定という制約があり、exactly.once.source.supportの既定はdisabledです。新規クラスタならenabledを直接指定すれば済みますが、稼働中のクラスタでは全ワーカーをpreparingにするローリング更新を1回、その後enabledにする更新をもう1回、合計2回まわします。コネクタ実装が対応していなければ設定だけでは効かないため、対応可否を先に確認してください。

リバランス挙動とtasks.max調整によるスループット拡張の勘所

ワーカーを増やせば速くなる、という理解だと運用で足をすくわれます。再配置の契機と並列度の上限は、別々の要因で決まるからです。

増分協調リバランスと5分の再配置遅延が運用計画に与える影響の把握

リバランスが起きるのは、コネクタの新規投入、タスク数の増減、設定変更、ワーカーの参加や離脱のときです。2.3.0以降の既定は増分協調リバランスで、移動が必要なタスクだけを止めます。全停止型に戻すconnect.protocolのeagerという値も残っていますが、戻す理由はほぼありません。

運用計画を立てる際に効くのはscheduled.rebalance.max.delay.msで、既定値は300000ミリ秒です。ワーカーが落ちてもこの5分間は再配置が始まらず、担当タスクは未割り当てのまま止まります。裏を返せば、デプロイ1台あたりの停止時間を5分の内側に収めれば、無用なタスク移動を避けられます。

シンク側のタスク数がパーティション数を超えても速度が伸びない理由

シンクコネクタのタスクは、内部的にはコンシューマグループのメンバーです。割り当てはパーティション単位で行われるため、タスク数がパーティション数を超えた分はレコードを受け取れません。tasks.maxに10を指定しても入力トピックが4パーティションなら、実質的な並列度は4で頭打ちになります。

スループットが出ないとき最初に見るのは、コネクタの設定値ではなく入力トピックのパーティション数です。ソース側の並列度はコネクタ実装が決めるため、テーブル単位でしかタスクを割らない実装ならテーブル数がそのまま上限になります。

Kafka Connectを採用しない方が総コストで有利になる3つの条件

そもそも使わない判断が正しい案件は珍しくありません。規模と遅延要件と体制の3点で線を引けます。

対象テーブル数と反映遅延の要件から引く採用ラインの具体的な基準

採用に踏み切る線は、連携対象が3系統以上あり、かつ秒から分の反映遅延に業務上の意味がある案件です。1系統だけで日次バッチの遅延が許されるなら見合いません。Connectワーカーは1台でもJVMを常時動かし、内部トピック3本を持ち、Java 17以上の環境を維持する構成だからです。

失敗しやすいのは「将来Kafkaを使うかもしれないから先に入れておく」という判断です。クラスタが無い状態でConnectだけを導入すると、ブローカーの運用コストが丸ごと上乗せされます。Kafkaがすでに業務データの通り道になっていること。これを満たさない案件では採用しないでください。

マネージドサービスへ寄せた方が総所有コストで有利になる3場面

次の3つに当てはまるならELTのマネージドサービスやOSSツールの方が総所有コストは下がります。連携元がSaaSのAPIばかりでコネクタの自前実装が要る場合。データ量が日次で数十万レコードにとどまりバッチ1回で捌ける場合。JVMプロセスの監視と復旧を担える運用体制が無い場合です。

該当する案件では、コネクタが数多く用意されたAirbyteのようなOSS版ELTツールに寄せた方が、立ち上がりも保守も軽く収まります。逆に既存のKafkaクラスタへリアルタイムで書き込む要件があるなら、この判断は覆ります。境界線はKafkaの有無です。

受託開発でKafka Connect基盤を立ち上げる際の分担と支援範囲

工数がかさむのはコネクタ設定そのものではありません。内部トピックの設計、DLQの監視と再投入の手順、スキーマ変更時の互換モードの取り決め、ワーカーの更新手順です。設定JSONは半日で書けても、これらの運用設計には数週間かかります。

自社では、Kafkaを含むデータ基盤の設計から構築・運用移管までをデータ分析基盤構築・MLOps構築支援として請け負う体制です。既存クラスタにConnectを足す部分だけを切り出す進め方も、基盤の設計から通しで担う進め方も選べます。社内にJVM運用の知見が無い段階でも、監視項目と復旧手順を引き渡し範囲に含めれば内製へ移せます。

よくある質問

Kafka Connectの導入検討でよく挙がる質問をまとめました。

Kafka Connectは別途インストールが必要ですか？

フレームワーク本体はApache Kafkaの配布物に同梱されているため、追加のインストールは不要です。起動スクリプトもKafkaのbinディレクトリに含まれています。ただしコネクタのプラグインは別で、同梱されているのは検証用のFileStreamSourceConnectorとFileStreamSinkConnectorのみです。実務で使うコネクタはJARを取得してplugin.path配下に置き、ワーカーを再起動して読み込ませます。

スタンドアロンモードは本番環境で使えますか？

推奨できません。単一プロセスで動くためフォールトトレランスが無く、プロセスが落ちれば連携も止まります。オフセットもローカルファイルに置かれるので、サーバ障害で読み位置ごと失う恐れがあります。加えてソースコネクタのexactly-onceは分散モード限定です。1台のログ収集のようにワーカーが1つで完結する用途を除き、本番は分散モードを選んでください。

Kafka Connectの実行にはどのJavaバージョンが必要ですか？

Kafka 4.0.0以降のConnectはJava 17以上が必要です。公式のJDK互換表では、ClientsとStreamsがJava 11でも動く一方、ConnectとServerはJava 11が非対応と明記されています。Java 8はKafka 4.0で削除されました。既存環境がJava 11のままなら、Connectワーカーを載せる前にランタイム更新を計画に入れてください。

コネクタが動いているのにデータが流れないときは何を見ますか？

まずステータスAPIでタスク単位の状態を確認します。Connectは失敗タスクを自動再起動しないため、コネクタがRUNNINGでもタスクだけFAILEDという状態がありえます。タスクが正常なら、次はワーカーのログでエラー許容の設定を疑ってください。errors.toleranceがallでDLQ未設定の場合、失敗レコードは黙って捨てられます。DLQトピックの件数も併せて見ます。

Kafka ConnectとKafka Streamsはどう使い分けますか？

Kafka Connectは外部システムとKafkaの間の入出力を担い、Kafka Streamsはトピック間の変換や集計を担います。SMTで書けるのはレコード1件ごとの軽い加工までです。取り込みと書き出しはConnect、業務ロジックを伴う変換はStreamsという分担にすると、責務が混ざらず障害の切り分けも楽になります。

関連記事

• データパイプラインとは：冪等性の観点からパイプライン全体を設計する視点をまとめています
• ストリーム処理とは：取り込んだデータをどう処理するか、バッチとの使い分けを解説しています
• ETLとは：ELTとの違いとツール選定の観点から、Connect以外の選択肢を整理しています
• データ分析基盤の構築とは：5層アーキテクチャの中でConnectが担う収集層の位置づけが分かります