コンシューマーラグのモニタリング¶

このドキュメントで説明する方法を使用して、Confluent Cloud でコンシューマーラグをモニタリングすることができます。

参考

Apache Kafka® クライアントアプリケーションおよび Confluent Cloud メトリクスをモニタリングする方法の例と、さまざまな失敗のシナリオに対応してメトリクスの結果を表示する手順については、Confluent Cloud に接続する Apache Kafka® クライアント用の Observability のデモを参照してください。

Metrics API 経由のコンシューマーラグのモニタリング¶

時間の経過に伴うコンシューマーラグの変化をモニタリングするには、Metrics API を使用することを推奨しています。トピックおよびコンシューマーグループの詳細レベルでモニタリングするには、サポートされている統合を使用できます。より詳細なコンシューマーおよびパーティションのレベルのデータを表示するには、サンプルクエリから始めることができます。

Note that this metric differs from the output of the kafka-consumer-groups command output in two ways.

Rebalancing Consumer Groups returns the last value for lag from before rebalancing until it completes.
Consumer Groups with no active Consumers are not returned in the result.

For example, a Consumer Group has two Consumers that are active but lagging behind by 3 and 4 records each. The Metrics API returns values of 3 and 4 for those Consumers respectively. A rebalance begins and more records are sent to the topic but they cannot be consumed until the rebalance completes. The rebalance takes two minutes. For those two minutes, the value for lag is 3 and 4 despite 20 records being sent to the topic during those two minutes. Upon completion of the rebalance, the Metrics API will return values of 13 and 14 for those Consumers respectively. Now both Consumers are turned off completely such that they no longer send their heartbeats. From this point, the Metrics API will no longer return data for this Consumer Group at all.

Confluent Cloud インターフェイス経由のコンシューマーラグのモニタリング¶

クラスター名を選択します。
Clients を選択し、Consumer Lag タブを選択します。コンシューマーグループのリストが表示されます。
リストからコンシューマーグループを選択して、そのグループのラグ詳細を表示します。

コンシューマーの作成について詳しくは、「Quick Start for Confluent Cloud」を参照してください。

Java クライアントメトリクス経由のオフセットラグのモニタリング¶

Java コンシューマーメトリクスをモニタリングすることができます。

クライアントインターセプター経由のコンシューマーレイテンシのモニタリング¶

Confluent Control Center を使用してコンシューマーレイテンシなどを追跡できます。

Kafka Admin API 経由のオフセットラグのモニタリング¶

Kafka Admin API および関連する CLI を使用してオフセットをモニタリングできます。この方法ではプログラムを使用してラグ情報にアクセスできます。詳細については、「AdminClient の構成」を参照してください。

前提条件

Confluent Cloud へのアクセス。
Java バージョン 1.7.0_111 以上、1.8.0_102 以上、および 1.9
ウェブブラウザー
Confluent Platform がインストールされていること

Confluent Cloud 構成を保持するクライアントプロパティファイルを作成します。この例では、client_ssl.properties という名前のファイルです。このファイルに Confluent Cloud クライアント構成を含める必要があります。この情報は、Confluent Cloud Console の CLI and client configuration タブで確認できます。この例を環境に合わせて構成します。
```
ssl.endpoint.identification.algorithm=https
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="<cloud-api-key>" \
password="<cloud-api-key-secret>";
security.protocol=SASL_SSL
```
BOOTSTRAP_SERVERS 変数を Confluent Cloud のクラスターブートストラップ URL に設定します。この値は、Cloud Console インターフェイスの Cluster settings をクリックして確認できます。
```
BOOTSTRAP_SERVERS="<bootstrap-url>"
```
Confluent Platform インストールホームから、コンシューマーグループをリスト表示します。--command-config 引数を使用して Confluent Cloud のプロパティを渡します。このスクリプトにはブートストラップサーバーを指定する必要があります。
```
./bin/kafka-consumer-groups --bootstrap-server ${BOOTSTRAP_SERVERS} --command-config \
client-ssl.properties --list  _confluent-healthcheck  example-group
```

コンシューマーグループごとに、このコマンドを使用してオフセットを確認します。このコマンドでは、Java コンシューマー API を使用しているコンシューマー（すなわち、ZooKeeper ベースでないコンシューマー）に関する情報のみが表示されます。

./bin/kafka-consumer-groups --bootstrap-server ${BOOTSTRAP_SERVERS} \
 --command-config /tmp/client.properties --describe --group _confluent-healthcheck

出力は以下のようになります。

TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
_confluent-healthcheck         0          13164704        13164773        69         healthcheck-agent-bf8d1655-63a6-4061-b680-0f11cdf182e5/100.96.67.0                   healthcheck-agent
_confluent-healthcheck         1          13161581        13161650        69         healthcheck-agent-bf8d1655-63a6-4061-b680-0f11cdf182e5/100.96.67.0                   healthcheck-agent
_confluent-healthcheck         2          12229509        12229578        69         healthcheck-agent-bf8d1655-63a6-4061-b680-0f11cdf182e5/100.96.67.0                   healthcheck-agent
_confluent-healthcheck         3          86              86              0          healthcheck-agent-bf8d1655-63a6-4061-b680-0f11cdf182e5/100.96.67.0                   healthcheck-agent
...

4 列目がラグ（最後にコミットされたオフセットと最新のログのオフセットの間の差分）を示しています。

kafka-consumer-groups の既知の問題¶

次のエラーが発生した場合:

Error: Executing consumer group command failed due to Failed to construct kafka consumer

または、TimeoutException でコマンドが失敗した場合は、client_ssl.properties の request.timeout.ms の構成（kafka-consumer-groups コマンドラインツールではデフォルトで 5000）を引き上げる必要があります。例を次に示します。

request.timeout.ms=60000