Kafka Source Connectors 実践ガイド: 外部システムからの取り込み設計とCCDAK対策

Kafka ConnectのSource Connectorは、データベースやファイル、SaaSなど外部システムからKafkaトピックへ連続的にデータを取り込む公式メカニズムです。

本稿では、アーキテクチャ、スケーリング、スキーマ/順序性、信頼性・運用、デプロイ選択、セキュリティ/性能チューニングを5〜7セクションで体系化し、CCDAK対策ポイントも明確化します。

Source Connectorの基礎と使いどころ

Kafka Connectはプラガブルなコネクタで外部システムとKafka間のデータ連携を標準化します。Source Connectorは外部からKafkaへの一方向同期を担い、ジョブの定義(Connector)と並列実行単位(Tasks)で構成されます。設定・実行・監視はREST APIから行えます。

実務では「安定運用」「再利用性」「メンテ性」が理由で、個別のProducerアプリより先にSource Connectorを検討します。特にRDB取り込み(JDBC/CDC)、オブジェクトストレージ、ログ/ファイル取り込み、SaaSイベント購読などで効果的です。

• 配信保証: 一般に少なくとも1回(at-least-once)。重複許容設計が前提。ConnectorやKafkaバージョンにより正確に一度を近似できる場合があるが、業務要件に合わせて重複排除や冪等設計を併用する。
• 内部状態: オフセットはKafkaの内部トピック(__connect-offsets)で管理。構成(__connect-configs)、ステータス(__connect-status)も内部トピックに保存。
• 軽度の変換はSMT(Single Message Transform)を優先。重い集約/結合が必要ならKafka StreamsやksqlDBを検討。

アーキテクチャとスケーリング: Workers・Connectors・Tasks

Kafka ConnectはWorkers(プロセス)がクラスタを形成し、Connector定義を分散管理、Tasksを各Workerに割り当てます。スケールアウトは主にtasks.maxの増加と、十分なKafkaパーティション数の確保で行います。再平衡はWorkerの増減や設定変更で発生します。

信頼性は内部トピックのレプリケーション・クォーラム、acks=all、十分なmin.insync.replicasで担保します。トピック設計とプロデューサ設定はConnectorから委譲/上書きされるため、要件に合わせて明示します。

• Workerモード: スタンドアロン(単機)か分散(クラスタ)。本番は分散が基本。
• Tasks並列度: tasks.maxに上限。実効並列は外部ソースの分割能力(例: テーブルシャーディング)やKafkaトピックのパーティション数にも依存。
• 内部トピック: __connect-configs/__connect-offsets/__connect-status は十分なreplication.factor(例: 3)を推奨。

Source Connectorの論理構成

取り込み設計: スキーマ、キー、パーティショニング、順序性

キー設計は最重要です。エンティティ単位の順序保証が必要なら、同一エンティティで同一キーを用い、同一パーティションに集約します。Kafkaの順序保証はパーティション内のみです。取り込み後の集約や下流処理のシャッフルも見越して、キーの安定性とカーディナリティを検討します。

スキーマはSchema Registryの利用を前提に設計すると変更管理が容易です。後方互換(Backward)を基本とし、列追加を優先、既存フィールドの削除/型変更は避けます。JDBC/CDCではDDL差分やNULL可否も考慮してください。

• 重複の前提: at-least-onceにより同一レコードの重複が生じ得る。キー+値のバージョンやソース側の増分カラムで冪等消込できる設計が望ましい。
• パーティション数: 消費スループットと順序要件のバランスで決定。エンティティ単位の直列性が必要な範囲がパーティション境界。
• SMTの活用: キー抽出、ヘッダ付与、ルーティング、フィールドマスクなどはSMTで軽量実装。

JDBC Source Connectorの例(増分同期)

{
  "name": "jdbc-users-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:postgresql://db:5432/app",
    "connection.user": "appuser",
    "connection.password": "${file:/opt/connect/secrets.properties:db.password}",
    "mode": "timestamp+incrementing",
    "incrementing.column.name": "id",
    "timestamp.column.name": "updated_at",
    "table.whitelist": "public.users",
    "topic.prefix": "src.postgres.",
    "poll.interval.ms": "10000",
    "batch.max.rows": "5000",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "producer.override.acks": "all",
    "producer.override.enable.idempotence": "true",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "_dlq.src.postgres.users",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

信頼性・配信セマンティクス・エラーハンドリング

Kafka ConnectのSourceは一般に少なくとも1回の配信です。送信成功後にオフセットをコミットするため、クラッシュ復旧時に重複送信が起こり得ます。冪等なキー設計、下流での重複排除、またはソース側の変更検知列を活用して整合性を保ちます。

エラー処理は段階的に設計します。レコード変換やスキーマ不一致はDLQへ退避し、運用で是正。リトライ可能な一時エラーはバックオフ付きで再試行し、恒久エラーはDLQ+アラートに切り分けます。

• 代表的な設定: errors.tolerance=all, errors.deadletterqueue.topic.name, errors.retry.timeout, errors.retry.delay.max.ms
• Producer側の堅牢化: acks=all, enable.idempotence=true, delivery.timeout.msの見直し、圧縮(compression.type)でスループット/コスト最適化
• 内部トピックはmin.insync.replicasを明示し、ブローカー障害時の耐性を確保

デプロイ方式の選択と運用: スタンドアロン vs 分散

開発/検証ではスタンドアロン、本番は分散モードが基本です。分散モードではConnector設定は内部トピックで共有され、Worker間で自動的にTasksが再割当てされます。更新やスケールアウトはREST APIからオンラインで実施できます。

独自Producerを実装すべきケースは限定的です。コネクタでサポートしないAPI仕様や厳格な独自トランザクション制御が必要な場合に検討します。まずは既存コネクタとSMTの組合せで要件充足を試みます。

• ローリング更新: Workerを1台ずつ再起動。再平衡の影響最小化へtasks.maxやクールダウンを調整
• スケール: Worker台数とtasks.maxを増やす前に、トピックのパーティション数と外部ソースの分割戦略を見直す
• 監視: コネクタ/タスクの状態、送信レート、エラーレート、DLQ滞留、再平衡頻度、内部トピックのレプリケーション

セキュリティと性能チューニング

Kafka接続はSASL/SSLで保護し、最小権限のACLを設定します。外部ソースの認証情報はConfig Providerで間接参照(例: ファイル/環境変数/外部シークレットストア)し、平文を避けます。監査上、DLQや内部トピックも含めてデータ分類を行ってください。

性能はエンド・ツー・エンド最適化が必要です。タスク並列、外部ソースのページング/スキャン戦略、Connector固有のpoll/batch設定、Producerのバッチ/遅延、Kafkaトピックのパーティション/圧縮、ネットワークMTU等を一貫して調整します。

• 認可: コネクタが書き込む全トピックに対しWrite/Create/DescribeConfigs等、必要最小限のACLを付与
• Config Provider: ${file:...}, ${env:...} などでシークレット値を参照。リロードポリシーと権限を管理
• チューニング例: poll.interval.ms、batch.max.rows(コネクタ依存)、producer.override.linger.ms/ batch.size、圧縮率とCPUのトレードオフ

問題で確認

よくある質問