ApsaraMQ for Kafka におけるコンシューマーの頻繁なリバランスのトラブルシューティング - ApsaraMQ for Kafka

コンシューマーの頻繁なリバランスは通常、メッセージ処理の遅延、コンシューマーパラメーターの設定ミス、または古いクライアントバージョンが原因で発生します。根本原因と修正方法はクライアントのバージョンによって異なります。

現象

ApsaraMQ for Kafka を使用していると、コンシューマークライアントでリバランスが頻繁に発生します。

原因

ハートビートのメカニズムがバージョンによって異なるため、根本原因はクライアントのバージョンによって異なります。

バージョン 0.10.1.0 より前のクライアント：専用のハートビートスレッドが存在しません。ハートビートは poll() の呼び出し内で送信されます。メッセージ処理に session.timeout.ms を超える時間がかかると、ブローカーはハートビートを受信せず、コンシューマーをグループから削除し、リバランスをトリガーします。
バージョン 0.10.1.0 以降のクライアント：専用のハートビートスレッドがメッセージ処理とは独立して実行されます。しかし、コンシューマーが max.poll.interval.ms で設定された間隔内に poll() を呼び出さない場合、クライアントはコンシューマーグループから離脱し、リバランスがトリガーされます。max.poll.interval.ms のデフォルト値は 5 分です。これは通常、メッセージのバッチ処理に時間がかかりすぎる場合に発生します。

リバランスに関連して、以下のエラーメッセージがよく見られます。発生しているエラーを特定することで、根本原因をより迅速に見つけるのに役立ちます：

NOT_COORDINATOR / join group failed：グループコーディネーターが変更された (例：ブローカーのフェールオーバーによる) か、ハートビートタイムアウトによりコンシューマーがグループから削除されたことを示します。コンシューマーはグループへの再参加を試み、リバランスをトリガーします。
MemberIdRequiredException / consumer poll timeout：バージョン 0.10.1.0 より前のクライアントの場合、これは通常、ハートビートをブロックするコンシューマーの処理停止によって引き起こされます。バージョン 0.10.1.0 以降のクライアントの場合、これはメッセージ処理の遅延により、コンシューマーが poll() を呼び出さずに max.poll.interval.ms を超えたことを示し、クライアントがコンシューマーグループから離脱する原因となります。
CommitFailedException：コンシューマーがメッセージの処理に時間がかかりすぎたか、時間内にハートビートを送信できなかったことを示します。ブローカーはオフセットコミットが完了する前にコンシューマーをグループから削除し (リバランスをトリガー) します。

主要パラメーター

以下のパラメーターがリバランスの動作を制御します：

パラメーター	適用バージョン	説明
`session.timeout.ms`	すべて	セッションタイムアウト。この期間内にハートビートが受信されない場合、ブローカーはコンシューマーをコンシューマーグループから削除します。
`max.poll.records`	すべて	`poll()` メソッドの呼び出しごとに返されるメッセージの最大数。
`max.poll.interval.ms`	0.10.1.0 以降	`poll()` メソッドの連続する 2 回の呼び出し間の最大間隔。この間隔を超えると、コンシューマーはコンシューマーグループから離脱し、リバランスがトリガーされます。
`heartbeat.interval.ms`	0.10.1.0 以降	コンシューマーがブローカーにハートビートを送信する間隔。`session.timeout.ms` 未満である必要があります。10,000 ミリ秒 (10 秒) 以下の値を推奨します。

解決策

パラメーター値の調整

クライアントのバージョンに基づいて、以下のパラメーターを設定します：

session.timeout.ms

クライアントバージョン	推奨値
0.10.1.0 より前	メッセージバッチの処理時間よりも長く、30 秒以下に設定します。25 秒を推奨します。
0.10.1.0 以降	デフォルト値の 10 秒を維持します。

max.poll.records

この値を次の数式の結果より十分に小さい値に設定します：

max.poll.records << messages_per_thread_per_second * number_of_threads * max.poll.interval.ms

max.poll.interval.ms (バージョン 0.10.1.0 以降のみ)

この値を次の数式の結果より大きい値に設定します：

max.poll.interval.ms > max.poll.records / (messages_per_thread_per_second * number_of_threads)

heartbeat.interval.ms (バージョン 0.10.1.0 以降)

heartbeat.interval.ms を 10,000 ミリ秒 (10 秒) 以下に設定します。この値は、コンシューマーがブローカーにハートビートを送信する頻度を制御し、常に session.timeout.ms より小さくする必要があります。一般的には、session.timeout.ms の 3 分の 1 に設定します。

Spring Kafka でのパラメーター設定

Spring Kafka を使用している場合、上記のすべてのパラメーターを application.yml ファイル内で設定できます：

spring:
  kafka:
    consumer:
      properties:
        session.timeout.ms: 10000
        heartbeat.interval.ms: 3000
        max.poll.interval.ms: 300000
        max.poll.records: 50

消費スループットに合わせて値を調整してください。max.poll.interval.ms の値は、max.poll.records のメッセージの単一バッチを処理するために必要な時間を超える必要があります。

消費速度の向上と処理スレッドの分離

メッセージ処理を専用のスレッドにオフロードし、コンシューマースレッドがスケジュールどおりに poll() を呼び出せるようにします。これにより、処理の遅延がハートビートをブロックしたり、ポーリング間隔を超えたりするのを防ぎます。
コンシューマーグループあたりのトピック数の削減

各コンシューマーグループがサブスクライブするトピックを 5 つ以下に維持します。最適な安定性を得るには、コンシューマーグループごとに 1 つのトピックをサブスクライブします。
バージョン 0.10.1.0 以降へのアップグレード

バージョン 0.10.1.0 より前のクライアントには専用のハートビートスレッドがないため、高負荷の処理中にハートビートタイムアウトが発生しやすくなります。バージョン 0.10.1.0 以降にアップグレードすると、ハートビート送信とメッセージ処理が分離され、この種のリバランスが解消されます。

よくある質問

Q：滞留メッセージの消費を高速化するために、リバランスを一時的に減らすにはどうすればよいですか？

リバランスを最小限に抑え、メッセージの滞留をできるだけ早く解消するには、次の手順を実行します：

クライアントバージョン 0.10.1.0 以降へのアップグレードバージョン 0.10.1.0 より前のクライアントには専用のハートビートスレッドがなく、高負荷下でリバランスが発生しがちです。他のパラメーターを調整する前にアップグレードしてください。
max.poll.interval.ms の増加メッセージの単一バッチを処理するために必要な最大時間よりも大きい値に設定してください。これにより、高負荷の処理中にコンシューマーが応答不能と見なされるのを防ぎます。
max.poll.records の削減poll() の呼び出しごとに取得されるメッセージの数を減らしてください。バッチが小さいほど処理が速く完了し、max.poll.interval.ms を超えるリスクが低減します。
コンシューマーグループあたりのサブスクライブトピック数の削減サブスクリプションを 5 つ以下のトピックに維持してください。最適な安定性を得るには、コンシューマーグループごとに 1 つのトピックをサブスクライブしてください。
コンシューマースレッドのブロッキング回避メッセージ処理ロジックを別の非同期スレッドに移動して、コンシューマースレッドがスケジュールどおりに poll() を呼び出せるようにしてください。