■ はじめに

Kafka Connect で 期待した通りのパフォーマンスがでないので
どう調査するのだろうって調べてた過程で、Rest API を見つけたので
メモしておく。

【１】Kafka connect の Rest API 一覧
　１）おさらい：curlコマンド
【２】コネクタ
　１）コネクタの登録
　２）コネクタ一覧表示
　３）コネクタの状況を確認
　４）コネクタの構成/タスク/タイプ
　５）コネクタの削除
【３】ログ
　１）ログレベルの確認
　２）ログレベルの変更

【１】Kafka connect の Rest API 一覧

* 以下に全てのRest API 一覧を見ることができる

https://docs.confluent.io/platform/7.1/connect/references/restapi.html
https://docs.confluent.io/ja-jp/platform/7.1/connect/monitoring.html

１）おさらい：curlコマンド

* curlコマンド の詳細については、以下の関連記事を参照のこと

curl コマンド
https://dk521123.hatenablog.com/entry/2017/12/05/233100

【２】コネクタ

１）コネクタの登録

curl -X POST \
  -H "Content-Type: application/json" \
  --data '{ "name": "demo-jdbc-sink", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "tasks.max": 1, "connection.url": "jdbc:postgresql://localhost:5431/demo_db", "connection.user": "postgres", "connection.password": "password", "insert.mode": "insert", "auto.create": "true", "topics": "demo_counter" } }' \
  http://localhost:8083/connectors

２）コネクタ一覧表示

curl -s http://localhost:8083/connectors/

# -s, --silent: 進捗やエラーを表示しない

３）コネクタの状況を確認

# curl -s http://localhost:8083/connectors/[CONNECT_NAME]/status/
curl -s http://localhost:8083/connectors/demo-jdbc-sink/status/

# -s, --silent: 進捗やエラーを表示しない

４）コネクタの構成/タスク/タイプ

# curl localhost:8083/connectors/[CONNECT_NAME]/tasks
curl localhost:8083/connectors/demo-jdbc-sink/tasks | jq

出力例

[
  {
    "id": {
      "connector": "demo-jdbc-sink",
      "task": 0
    },
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
      "connection.password": "password",
      "tasks.max": "1",
      "topics": "demo_counter",
      "key.converter.schemas.enable": "false",
      "delete.enabled": "true",
      "auto.evolve": "true",
      "task.class": "io.confluent.connect.jdbc.sink.JdbcSinkTask",
      "connection.user": "postgres",
      "value.converter.schemas.enable": "false",
      "name": "demo-jdbc-sink",
      "auto.create": "true",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "connection.url": "jdbc:postgresql://postgresql:5432/demo_db",
      "insert.mode": "upsert",
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
      "pk.mode": "record_key",
      "pk.fields": "word"
    }
  }
]

構成のみ

# curl localhost:8083/connectors/[CONNECT_NAME]/config
curl localhost:8083/connectors/demo-jdbc-sink/config | jq

５）コネクタの削除

https://docs.confluent.io/platform/7.1/connect/references/restapi.html#delete--connectors-(string-name)-

# DELETE /connectors/(string:name)/
curl -X DELETE -s http://localhost:8083/connectors/demo-jdbc-sink

【３】ログ

https://developer.confluent.io/learn-kafka/kafka-connect/troubleshooting-kafka-connect/#dynamic-log-configuration

１）ログレベルの確認

curl -s http://localhost:8083/admin/logger/
# -s, --silent: 進捗やエラーを表示しない

# jq がインストールされている場合
curl -s http://localhost:8083/admin/logger/ | jq

https://docs.confluent.io/platform/7.1/connect/logging.html#check-log-levels

２）ログレベルの変更

curl -s -X PUT -H "Content-Type:application/json" \
http://localhost:8083/admin/loggers/org.apache.kafka.connect.runtime.WorkerSourceTask \
-d '{"level": "TRACE"}' | jq '.'

https://docs.confluent.io/platform/7.1/connect/logging.html#change-the-log-level-for-a-specific-logger
補足

# -X: HTTPメソッド(GET/POST/PUT/DELETE etc)を指定
# -H: Request Headerを追加（Content-Typeを指定する場合）
# -d: POST でフォームを送信する。引数で送信するコンテンツを指定する

出力結果

[
  "org.apache.kafka.connect.runtime.WorkerSourceTask"
]

参考文献

https://developer.confluent.io/learn-kafka/kafka-connect/troubleshooting-kafka-connect/
https://docs.confluent.io/ja-jp/platform/7.1/connect/monitoring.html

プログラムの超個人的なメモ

Memo for Programming.

【Kafka】Kafka Connect ～ Rest API ～

■ はじめに

目次

【１】Kafka connect の Rest API 一覧

１）おさらい：curlコマンド

【２】コネクタ

１）コネクタの登録

２）コネクタ一覧表示

３）コネクタの状況を確認

４）コネクタの構成/タスク/タイプ

５）コネクタの削除

【３】ログ

１）ログレベルの確認

２）ログレベルの変更

参考文献

関連記事