Kafka Connect REST APIで実現するコネクタ管理とCI/CD連携

Kafka Connectは分散ワーカー上でコネクタを実行し、その制御面(API)としてRESTを提供します。CI/CDから宣言的に操作できる点が強みです。

本稿は公式ドキュメントの安定機能に基づき、コネクタ作成・更新・停止、ステータス監視、オフセット操作、検証APIの使い分けを具体的に解説します。

Connect REST APIの全体像とデプロイ前提

Kafka ConnectのREST APIは各ワーカーでリッスンし、/connectorsや/connector-plugins等のエンドポイントを介してコネクタを宣言的に管理します。分散モードではコネクタ設定・タスク・ステータスが内部トピックに保存されるため、ワーカー障害時も状態はKafkaに残ります。

CI/CD連携では、POSTで新規作成、PUTで構成の宣言的更新(基本的に冪等)を行います。更新により該当コネクタのタスクは再起動されるため、変更が無停止で反映されるわけではありませんが、通常は短時間の再均衡で収まります。

• 典型ポート: 8083 (listenersで変更可)。TLS/リバースプロキシでの保護が一般的。
• 分散モード前提: config.storage.topic / offset.storage.topic / status.storage.topic を適切にプロビジョニング。
• 実務/試験の要点: 作成はPOST、更新はPUT、状態はGET /status、タスク単位の再起動はPOST /tasks/{id}/restart。

CI/CDからConnect RESTへのフロー(概念図)

基本の一覧・存在確認(例)

# コネクタ一覧
curl -s http://connect.example.com:8083/connectors | jq .

# 個別ステータス
curl -s http://connect.example.com:8083/connectors/my-sink/status | jq .

# プラグイン一覧
curl -s http://connect.example.com:8083/connector-plugins | jq '.[] | .class'

コネクタのライフサイクル: 作成・更新・一時停止・再起動・削除

新規作成はPOST /connectorsにnameとconfigを送ります。すでに同名が存在する場合は409系応答が一般的です。宣言的更新はPUT /connectors/{name}/configで行い、CI/CDではこのPUTを中心に据えることで冪等なデプロイが可能になります。

一時停止/再開はパイプラインの安全な変更や下流メンテ時に有用です。再起動は失敗回復や構成変更反映に使用します。削除は内部状態も含めてクリーンに落としますが、オフセットは別管理(後述)です。

• 試験では「POST=新規、PUT=更新(冪等)」の区別が頻出。
• PUT更新は通常、該当コネクタのタスク再起動を伴う点に注意(短時間のブリップを許容する設計に)。
• pause/resumeはデータフローの一時停止/再開。deleteは構成を削除し、復旧には再登録が必要。

更新時のシーケンス(簡略)

作成と冪等更新の例

# 新規作成(POST)
curl -s -X POST http://connect:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{
        "name": "orders-sink",
        "config": {
          "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
          "topics": "orders",
          "connection.url": "jdbc:postgresql://db/orders",
          "auto.create": "true"
        }
      }' | jq .

# 宣言的更新(PUT) - 同じボディをPUTすれば冪等
curl -s -X PUT http://connect:8083/connectors/orders-sink/config \
  -H 'Content-Type: application/json' \
  -d '{
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "topics": "orders",
        "connection.url": "jdbc:postgresql://db/orders",
        "auto.create": "true",
        "insert.mode": "upsert"
      }' | jq .

# 一時停止・再開・再起動
curl -s -X PUT  http://connect:8083/connectors/orders-sink/pause
curl -s -X PUT  http://connect:8083/connectors/orders-sink/resume
curl -s -X POST http://connect:8083/connectors/orders-sink/restart

オフセット管理・タスク制御・エラー対処

オフセットは内部トピックに保持され、RESTから参照・リセットが可能です。近年のKafka Connectでは/connectors/{name}/offsetsで参照、DELETEでリセット(一部/全体)がサポートされます。部分リセットは対象パーティション(もしくはソース側パーティション表現)をボディで指定します。

FAILEDタスクはPOST /connectors/{name}/tasks/{id}/restartで個別再起動できます。エラー処理はerrors.toleranceやDead Letter Queue(DLQ)関連設定で制御し、致命的エラーはFAILEDとなるため、ステータス監視と組み合わせて自動回復を設計します。

• オフセットのリセットは再取り込み・再配送を引き起こすため、本番では明示的に範囲を限定。
• FAILEDの原因はstatus出力のtraceやワーカーのログ(JMX/ログ)で確認。
• DLQ設定(例: errors.deadletterqueue.topic.name)はシンク/ソースで挙動差あり。

エラー発生〜回復の流れ(概念)

ステータス/オフセット/再起動の例

# ステータス確認
curl -s http://connect:8083/connectors/orders-sink/status | jq .

# タスク0を再起動
curl -s -X POST http://connect:8083/connectors/orders-sink/tasks/0/restart

# オフセット参照
curl -s http://connect:8083/connectors/orders-sink/offsets | jq .

# オフセット部分リセット(例: パーティション指定)
curl -s -X DELETE http://connect:8083/connectors/orders-sink/offsets \
  -H 'Content-Type: application/json' \
  -d '{
        "partitions": [ { "topic": "orders", "partition": 0 } ]
      }'

CI/CD統合パターン: 宣言的運用とシークレット分離

Gitにコネクタ設定(JSON)を格納し、CIからPUTで適用するGitOpsが実務上安定です。環境差分は環境変数やテンプレートで埋め込み、同一の論理名を保つことでプロモーションを容易にします。

シークレットはConfigProvider(例: ${env:VAR}, ${file:/path:key} 等)や外部秘密管理プラグインを利用し、設定ファイルに平文を残さない方針が基本です。RBACやプロキシでREST面の認可を掛け、監査ログを残します。

• PUT中心の宣言的適用により、同じパイプラインをDev/Stage/Prodで再現可能。
• Blue/Green切替は-suffix(v2等)の並行稼働→切替→旧停止が安全。
• ConfigProviderを使うとRESTボディや構成トピックにも秘匿値を露出しにくい。

GitOps適用(簡略)

GitHub Actionsでの宣言的デプロイ例

name: deploy-connector
on: [push]
jobs:
  apply:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Render config
        run: |
          envsubst < config/orders-sink.json > rendered.json
      - name: Upsert via PUT
        run: |
          set -e
          CONNECT_BASE=${{ secrets.CONNECT_URL }}
          curl -sS -X PUT "$CONNECT_BASE/connectors/orders-sink/config" \
            -H 'Content-Type: application/json' \
            --data-binary @rendered.json | jq .
      - name: Verify status
        run: |
          CONNECT_BASE=${{ secrets.CONNECT_URL }}
          for i in {1..20}; do
            st=$(curl -s "$CONNECT_BASE/connectors/orders-sink/status" | jq -r .connector.state)
            [ "$st" = "RUNNING" ] && exit 0
            sleep 3
          done
          echo "Connector not RUNNING" >&2; exit 1

プラグイン検出と構成検証APIの活用

デプロイ前に/connector-pluginsで対象クラスが存在するかを確認します。plugin.pathの設定やクラスローダーの範囲により見え方が変わるため、CIの事前検証に組み込みます。

一部のバージョンでは構成検証エンドポイント(例: /connector-plugins/{class}/config/validate)が提供され、要求項目の欠落や型不一致をRESTで確認できます。HTTPメソッドは実装・バージョンで差異があるため、環境のドキュメントを必ず確認してください。

• configにはconnector.class(完全修飾クラス名)を必ず指定。nameは作成時の識別子で別物。
• 検証APIがない/差異がある環境では、ダミー作成を避けるために静的バリデーション(スキーマ化)を補助的に利用。
• シリアライズやスキーマ互換性(特にSchema Registry連携)はコネクタ外側の前提として別途検証。

検証→適用の流れ

プラグイン確認と検証(例。環境によってメソッド差異に注意)

# プラグイン一覧
curl -s http://connect:8083/connector-plugins | jq '.[] | .class'

# 構成検証(例: PUT/POSTは環境に従う)
CLASS=io.confluent.connect.jdbc.JdbcSinkConnector
curl -s -X PUT http://connect:8083/connector-plugins/$CLASS/config/validate \
  -H 'Content-Type: application/json' \
  -d '{
        "name": "validate-only",
        "config": {
          "connector.class": "'$CLASS'",
          "topics": "orders"
        }
      }' | jq .

運用監視とセキュリティ留意点

稼働確認は/statusのRUNNING/FAILEDと、ワーカーのJMXメトリクス(スループット、エラー率、タスク稼働数等)を併用します。ヘルスチェック用に軽量なGET /connectorsや特定コネクタの/statusを利用し、CI/CD後の検証ステップに組み込みます。

REST面はTLS終端と認証・認可で保護します。自己管理のKafka ConnectではHTTPの前段にリバースプロキシを置き、Basic認証やmTLS・IP制限を適用する構成が一般的です。Confluent Platform環境ではRBACを使った粒度の細かい制御も可能です。

• ロールリング更新時はタスク・ワーカーの再均衡に伴う一時的な遅延を監視。
• APIは失敗時に再試行(指数バックオフ)を実装。409/409相当は整合的ハンドリング。
• RESTはパブリックに晒さない。最小権限・監査ログ・証明書ローテーションを徹底。

監視の面(概念)

Status API -> Health
JMX -> Metrics
Logs -> Root cause

簡易ヘルスチェックとセキュアな呼び出し例

# 200かつJSONなら疎通OKとみなす
curl -fsS https://connect.example.com:8443/connectors \
  --cacert /etc/ssl/certs/ca.pem \
  -u ci-user:${CONNECT_REST_PASSWORD} | jq '.' >/dev/null

# 指数バックオフでステータス確認
for i in 1 2 3 5 8 13; do
  state=$(curl -s https://connect.example.com:8443/connectors/orders-sink/status \
    --cacert /etc/ssl/certs/ca.pem -u ci-user:${CONNECT_REST_PASSWORD} | jq -r .connector.state || echo "")
  [ "$state" = "RUNNING" ] && break
  sleep $i
done

問題で確認

よくある質問