Kafka Connect とコネクターのモニタリング¶
JMX および REST インターフェイスを使用して、Connect、コネクター、およびクライアントを管理およびモニタリングすることができます。
REST インターフェイスの使用¶
Kafka Connect の REST API を使用して、クラスターを管理することができます。これには、コネクターの構成とそのタスクのステータスを表示する API や、コネクターの現在の動作(構成の変更やタスクの再起動など)を変更する API が含まれます。
Kafka Connect クラスターが起動したら、モニタリングしたり変更したりすることができます。このセクションでは、REST API を使用して行われるいくつかの一般的な管理タスクについて説明します。
Kafka Connect は、サービスとして実行できるように設計されており、コネクターを管理するための REST API もサポートされています。デフォルトでは、このサービスはポート 8083
で実行されます。分散モードで実行する場合、REST API はクラスターに対するプライマリインターフェイスになります。リクエストは、どのクラスターメンバーに対しても行うことができます。REST API は必要に応じてリクエストを自動的に転送します。
コマンドラインでコネクターを送信するだけでスタンドアロンモードを使用することができますが、それによって REST インターフェイスも実行されます。これは、処理を停止せずにステータス情報を取得したり、コネクターを追加したり削除したりするのに便利です。
コネクターとタスクのステータス¶
REST API を使用して、コネクターとそのタスクの現在のステータスや、各コネクターが割り当てられているワーカーの ID を表示することができます。
コネクターとそのタスクは、クラスター内のすべてのワーカーがモニタリングする共有トピックに対してステータスの更新をパブリッシュします(この共有トピックは status.storage.topic
で構成されます)。ワーカーはこのトピックを非同期で消費するため、通常は、ステータス API でステートの変更を確認できるまでに短い遅延が発生します。コネクターまたはそのタスクの 1 つは、次のいずれかのステートになります。
UNASSIGNED
: コネクター/タスクは、まだワーカーに割り当てられていません。RUNNING
: コネクター/タスクは実行中です。PAUSED
: コネクター/タスクは、管理のために一時停止中です。FAILED:
コネクター/タスクが失敗しました(通常は、例外の発生によるもので、これはステータス出力で報告されます)。
ほとんどの場合、コネクターとタスクのステートは一致しますが、変更が生じていたりタスクが失敗したりした場合などは、一時的に不一致の状態になる可能性があります。たとえば、コネクターが最初に起動されてからコネクターとそのタスクがすべて RUNNING
ステートに遷移するまでには、大幅な遅延が発生することがあります。Connect では失敗したタスクが自動的に再起動されないため、タスクが失敗したときにもステートの不一致が生じます。
コネクターのメッセージ処理を一時的に停止することが役立つ場合があります。たとえば、リモートシステムがメンテナンス中の場合は、大量の例外でログが埋まることを避けるため、ソースコネクターでは新しいデータを確認するためのポーリングを停止することが望ましいでしょう。このユースケースに対応できるよう、Connect では一時停止/再開の API が提供されています。ソースコネクターが一時停止されている間、Connect は追加レコードの取得のためのポーリングを停止します。シンクコネクターが一時停止されている間、Connect はシンクコネクターへの新しいメッセージのプッシュを停止します。一時停止ステートは永続的であるため、クラスターを再起動しても、タスクが再開されるまでコネクターはメッセージ処理を再開しません。タスクが一時停止されたときに実行中だった処理が完了するまでに時間がかかる場合があるため、すべてのコネクターのタスクが PAUSED
ステートに移行するまでに遅延が生じる場合があることに注意してください。また、失敗したタスクは再起動されるまで PAUSED
ステートには遷移しません。
一般的な REST の例¶
以下に、REST API を介して行われるいくつかの一般的なアクティビティを紹介します。以下に示す例では、デフォルトの構成を使用して localhost 上で実行されているワーカーと s3-connector
という名前のコネクターを使用しています。この例では、応答のフォーマットにユーティリティー jq
が使用されていますが、これは必須ではありません。ここでは、例を意図的にシンプルにしています。REST API の高度な使い方については、 Kafka Connect REST インターフェイス を参照してください。
ワーカークラスター ID、バージョン、および git ソースコードのコミット ID を取得します。
curl localhost:8083/ | jq
出力例:
{ "version": "5.5.0-ce", "commit": "2bedb2c6980ba7a8", "kafka_cluster_id": "wxOOb2vrTPCLJl28PI18KA" }
ワーカーで使用可能なコネクタープラグインのリストを表示します。
curl localhost:8083/connector-plugins | jq
出力例:
[ { "class": "io.confluent.connect.activemq.ActiveMQSourceConnector", "type": "source", "version": "5.5.0" }, { "class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector", "type": "sink", "version": "5.5.0" }, { "class": "io.confluent.connect.ibm.mq.IbmMQSourceConnector", "type": "source", "version": "5.5.0" }, { "class": "io.confluent.connect.jdbc.JdbcSinkConnector", "type": "sink", "version": "5.5.0" }, { "class": "io.confluent.connect.jdbc.JdbcSourceConnector", "type": "source", "version": "5.5.0" }, { "class": "io.confluent.connect.jms.JmsSourceConnector", "type": "source", "version": "5.5.0" }, ... omitted
ワーカーのアクティブなコネクターのリストを表示します。
curl localhost:8083/connectors
出力例:
["s3-connector"]
コネクターを再起動します(コマンドが成功しても出力は生成されません)。
curl -X POST localhost:8083/connectors/s3-connector/restart
注釈
コネクターを再起動してもタスクは再開されません。
コネクターのタスクを取得します。
curl localhost:8083/connectors/s3-connector/tasks | jq
出力例:
[ { "id": { "connector": "s3-connector", "task": 0 }, "config": { "connector.class": "io.confluent.connect.s3.S3SinkConnector", "format.class": "io.confluent.connect.s3.format.bytearray.ByteArrayFormat", "task.class": "io.confluent.connect.s3.S3SinkTask", "flush.size": "1000", "topics": "passengers", "name": "s3-connector", "aws.access.key.id": "omitted", "rotate.interval.ms": "-3", "aws.secret.access.key": "omitted", "storage.class": "io.confluent.connect.s3.storage.S3Storage", "s3.bucket.name": "s3-test-bucket" } } ]
タスクを再起動します(コマンドが成功しても何も出力されません)。
curl -X POST localhost:8083/connectors/s3-connector/tasks/0/restart
コネクターを一時停止します(コマンドが成功しても何も出力されません)。
curl -X PUT localhost:8083/connectors/s3-connector/pause
ちなみに
コネクターとやり取りするシステムのサービスを、メンテナンスのために一時的に停止する必要がある場合に、このコマンドが役立ちます。
一時停止されているコネクターを再開します(コマンドが成功しても何も出力されません)。
curl -X PUT localhost:8083/connectors/s3-connector/resume
コネクターの構成を更新します。
curl -X PUT -H "Content-Type: application/json" --data '{"connector.class":"io.confluent.connect.s3.S3SinkConnector","format.class":"io.confluent.connect.s3.format.bytearray.ByteArrayFormat","flush.size":"1000","s3.bucket.name":"s3-test-bucket","storage.class":"io.confluent.connect.s3.storage.S3Storage","tasks.max":"2","topics":"passengers","name":"s3-connector"}' localhost:8083/connectors/s3-connector/config | jq
コマンド例では、タスクを 1 から 2 に更新しています。
{ "name": "s3-connector", "config": { "connector.class": "io.confluent.connect.s3.S3SinkConnector", "format.class": "io.confluent.connect.s3.format.bytearray.ByteArrayFormat", "flush.size": "1000", "s3.bucket.name": "s3-test-bucket", "storage.class": "io.confluent.connect.s3.storage.S3Storage", "tasks.max": "2", "topics": "passengers", "name": "s3-connector" }, "tasks": [ { "connector": "s3-connector", "task": 0 }, { "connector": "s3-connector", "task": 1 } ], "type": "sink" }
コネクターのステータスを取得します。
curl localhost:8083/connectors/s3-connector/status | jq
出力例:
{ "name": "s3-connector", "connector": { "state": "RUNNING", "worker_id": "192.168.86.66:8083" }, "tasks": [ { "id": 0, "state": "RUNNING", "worker_id": "192.168.86.66:8083" }, { "id": 1, "state": "RUNNING", "worker_id": "192.168.86.66:8083" } ], "type": "sink" }
コネクターの構成、タスク、およびコネクターのタイプを取得します。
curl localhost:8083/connectors/s3-connector | jq
出力例:
{ "name": "s3-connector", "config": { "connector.class": "io.confluent.connect.s3.S3SinkConnector", "format.class": "io.confluent.connect.s3.format.bytearray.ByteArrayFormat", "flush.size": "1000", "tasks.max": "2", "topics": "passengers", "name": "s3-connector", "storage.class": "io.confluent.connect.s3.storage.S3Storage", "s3.bucket.name": "s3-test-bucket" }, "tasks": [ { "connector": "s3-connector", "task": 0 }, { "connector": "s3-connector", "task": 1 } ], "type": "sink" }
コネクターの構成のみを取得するには、以下のコマンドを使用します。
curl localhost:8083/connectors/s3-connector/config | jq
コネクターを削除します(コマンドが成功しても何も出力されません)。
curl -X DELETE localhost:8083/connectors/s3-connector
JMX を使用した Connect のモニタリング¶
Connect では、JMX(Java Management Extensions)を通じてさまざまなメトリクスが報告されます。metric.reporters
構成オプションで、プラグ可能な統計情報レポーターを追加して、統計情報が報告されるように Connect を構成することができます。以下のセクションでは MBeans、メトリクス、および説明を示します。
コネクターのメトリクス¶
MBean: kafka.connect:type=connector-metrics,connector="{([-.w]+)
メトリック | 説明 |
---|---|
connector-type | コネクターのタイプ: ソースまたはシンク |
connector-class | コネクタークラスの名前 |
connector-version | コネクタークラスのバージョン(コネクターによって報告される) |
status | コネクターのステータス: running、paused、または stopped |
共通タスクのメトリクス¶
MBean: kafka.connect:type=connector-task-metrics,connector="{connector}",task="{task}"
メトリック | 説明 |
---|---|
status | 現在のタスクのステータス: unassigned、running、paused、failed、または destroyed |
pause-ratio | タスクが paused ステートで経過した時間の割合 |
running-ratio | タスクが running ステートで経過した時間の割合 |
offset-commit-success-percentage | タスクによるオフセットコミットの試行のうち、成功した試行の平均比率(パーセンテージ) |
offset-commit-failure-percentage | タスクによるオフセットのコミットの試行のうち、失敗またはエラーになった試行の平均比率(パーセンテージ) |
offset-commit-max-time-ms | タスクがオフセットをコミットする際にかかった最大時間(単位: ミリ秒) |
offset-commit-avg-time-ms | タスクがオフセットをコミットするためにかかった平均時間(単位: ミリ秒) |
batch-size-max | コネクターで処理されるバッチの最大サイズ |
batch-size-avg | コネクターで処理されるバッチの平均サイズ |
ワーカーのメトリクス¶
MBean: kafka.connect:type=connect-worker-metrics
メトリック | 説明 |
---|---|
task-count | このワーカーで実行されたタスクの数 |
connector-count | このワーカーで実行されたコネクターの数 |
connector-startup-attempts-total | このワーカーがコネクターの起動を試行した回数の合計 |
connector-startup-success-total | コネクターが正常に起動した回数の合計 |
connector-startup-success-percentage | ワーカーのコネクターの起動回数のうち、成功した回数の平均比率(パーセンテージ) |
connector-startup-failure-total | コネクターの起動に失敗した回数の合計 |
connector-startup-failure-percentage | ワーカーのコネクターの起動回数のうち、失敗した回数の平均比率(パーセンテージ) |
task-startup-attempts-total | ワーカーがタスクの起動を試行した回数の合計 |
task-startup-success-total | タスクの起動に成功した回数の合計 |
task-startup-success-percentage | ワーカーのタスクの起動回数のうち、成功した回数の平均比率(パーセンテージ) |
task-startup-failure-total | タスクの起動に失敗した回数の合計 |
task-startup-failure-percentage | ワーカーのタスクの起動回数のうち、失敗した回数の平均比率(パーセンテージ) |
ワーカーのバランス調整のメトリクス¶
MBean: kafka.connect:type=connect-worker-metrics
メトリック | 説明 |
---|---|
leader-name | グループリーダーの名前 |
epoch | ワーカーのエポックまたは世代番号 |
completed-rebalances-total | ワーカーによって実行されたバランス調整の合計数 |
rebalancing | ワーカーが現在バランス調整を行っているかどうか |
rebalance-max-time-ms | ワーカーがバランス調整に費やした最大時間(単位: ミリ秒) |
rebalance-avg-time-ms | ワーカーがバランス調整に費やした平均時間(単位: ミリ秒) |
time-since-last-rebalance-ms | 直近でワーカーのバランス調整が行われてから経過した時間(単位: ミリ秒) |
ソースタスクのメトリクス¶
MBean: kafka.connect:type=source-task-metrics,connector=([-.w]+),task=([d]+)
メトリック | 説明 |
---|---|
source-record-write-total | (タスクが最後に再開されて以降)変換によって出力され、ワーカー内の名前付きソースコネクターに属するタスクに対して Kafka に書き込まれたレコードの件数 |
source-record-write-rate | 変換の適用後に、変換によって出力され、ワーカー内の名前付きソースコネクターに属するタスクに対して Kafka に書き込まれたレコードの 1 秒あたりの平均件数(変換時のフィルターで除外されたレコードは除きます) |
source-record-poll-total | (タスクが最後に再開されて以降)変換の適用前に、ワーカー内の名前付きソースコネクターに属するタスクによって生成またはポーリングされたレコードの件数 |
source-record-poll-rate | 変換の適用前に、ワーカー内の名前付きソースコネクターに属するタスクによって生成またはポーリングされたレコードの 1 秒あたりの平均件数 |
source-record-active-count-max | Maximum number of records produced by the task but not yet completely written to Kafka |
source-record-active-count-avg | Average number of records produced by the task but not yet completely written to Kafka |
source-record-active-count | Most recent number of records produced by the task but not yet completely written to Kafka |
poll-batch-max-time-ms | このタスクでソースレコードのバッチをポーリングする際にかかった最大時間(単位: ミリ秒) |
poll-batch-avg-time-ms | このタスクでソースレコードのバッチをポーリングする際にかかった平均時間(単位: ミリ秒) |
シンクタスクのメトリクス¶
MBean: kafka.connect:type=sink-task-metrics,connector=([-.w]+),task=([d]+)
メトリック | 説明 |
---|---|
sink-record-read-rate | 変換の適用前に、ワーカー内の名前付きシンクコネクターに属するタスクに対して Kafka から読み取られたレコードの 1 秒あたりの平均件数 |
sink-record-read-total | Before transformations are applied, this is the total number of records produced or polled by the task belonging to the named source connector in the worker (since the task was last restarted) |
sink-record-send-rate | 変換の適用後に、変換によって出力され、ワーカー内の名前付きシンクコネクターに属するタスクに送信されたレコードの 1 秒あたりの平均件数(変換時のフィルターで除外されたレコードは除きます) |
sink-record-send-total | Total number of records output from the transformations and sent to the task belonging to the named source connector in the worker (since the task was last restarted) |
sink-record-active-count | シンクタスクが Kafka から読み取り、まだコミット、フラッシュ、または確認応答を完了していないレコードの最新の件数 |
sink-record-active-count-max | シンクタスクが Kafka から読み取り、まだコミット、フラッシュ、または確認応答を完了していないレコードの最大件数 |
sink-record-active-count-avg | シンクタスクが Kafka から読み取り、まだコミット、フラッシュ、または確認応答を完了していないレコードの平均件数 |
partition-count | ワーカー内の名前付きシンクコネクターに属する、タスクに割り当てられたトピックパーティションの数 |
offset-commit-seq-no | オフセットコミットの現在のシーケンス番号 |
offset-commit-completion-rate | 正常に完了したオフセットのコミットの 1 秒あたりの平均完了件数 |
offset-commit-completion-total | 正常に完了したオフセットのコミットの合計完了件数 |
offset-commit-skip-rate | 受信が遅すぎてスキップまたは無視されたオフセットコミットの 1 秒あたりの平均完了件数 |
offset-commit-skip-total | 受信が遅すぎてスキップまたは無視されたオフセットコミットの合計完了件数 |
put-batch-max-time-ms | このタスクでシンクレコードのバッチを配置する際にかかった最大時間(単位: ミリ秒) |
クライアントのメトリクス¶
MBean: kafka.connect:type=connect-metrics,client-id=
メトリック | 説明 |
---|---|
connection-close-rate | ウィンドウにおける、1 秒あたりの終了された接続の数 |
connection-count | 現在アクティブな接続の数 |
connection-creation-rate | ウィンドウにおける、1 秒あたりの新規接続の確立数 |
failed-authentication-rate | 認証に失敗した接続 |
incoming-byte-rate | すべてのソケットから読み取られた 1 秒あたりのバイト数 |
io-ratio | I/O スレッドで I/O 処理に費やされた時間の割合 |
io-time-ns-avg | select 呼び出しごとの I/O の所要時間の平均値(単位: ナノ秒) |
io-wait-ratio | I/O スレッドで待機に費やされた時間の割合 |
io-wait-time-ns-avg | ソケットで読み取りまたは書き込みの準備が整うまでの I/O スレッドの待機時間の平均値(単位: ナノ秒) |
network-io-rate | すべての接続における 1 秒あたりのネットワーク操作(読み取りまたは書き込み)の平均回数 |
outgoing-byte-rate | 1 秒あたりの平均送信バイト数(すべてのサーバーに対する送信) |
request-rate | 1 秒あたりの平均送信リクエスト数 |
request-size-avg | ウィンドウにおけるすべてのリクエストの平均サイズ |
request-size-max | ウィンドウにおける送信リクエストの最大サイズ |
response-rate | 1 秒あたりに送受信された応答 |
select-rate | I/O レイヤーで新規 I/O の実行がチェックされた、1 秒あたりの回数 |
successful-authentication-rate | SASL または SSL を使用して正常に認証された接続 |