解決策

Apache Kafka のコンシューマーは通常、コンシューマーグループに含まれています。コンシューマーグループの各コンシューマーは、次の場合にトピック内の異なるパーティションのサブセットからメッセージを受信します。

• 複数のコンシューマーがトピックを購読している。
• これらのコンシューマーが同じコンシューマーグループに属している。

コンシューマーがクラスターに到達できない場合、グループコーディネーターがコンシューマーをコンシューマーグループから削除します。このプロセスにより、以下のアクションを含む再調整イベントが開始します。

• 残りのコンシューマーはパーティションから解放されます。
• グループコーディネーターは、残りのコンシューマーにトピックのパーティションを再配布します。

次の状況では、コンシューマーグループが再調整を行う可能性があります。

• パーティションの所有権を、コンシューマーグループのあるコンシューマーから別のコンシューマーに移動しました。
• コンシューマーグループに新しいコンシューマーを追加しました。
• コンシューマーがコンシューマーグループからシャットダウン、クラッシュ、または脱退しました。
• トピックを変更しました。パーティションの再編成が行われます。
• コンシューマーグループのサブスクリプションにクライアント設定に関する問題があります。これは、グループがサブスクライブしているトピックと、グループ内の個々のユーザーに割り当てられたトピックが一致しないことが原因です。

同じコンシューマーグループ内のコンシューマーは、再調整イベントが完了するまではデータの消費を続行することができません。これはパーティション割り当てのデフォルト動作です。これは、パーティション割り当て戦略を CooperativeStickyAssignor に変更することで回避できます。

コンシューマーグループによる再調整が継続的に行われないようにするには、以下を試してください。

• コンシューマー設定で max.partition.fetch.bytes の値を小さくするか、セッションタイムアウト (session.timeout.ms) の値を増やします。セッションのタイムアウトとそれに続く再調整を避けるために、コンシューマーは poll() を頻繁に呼び出す必要があります。1 つの poll() が返すデータ量が多い場合、コンシューマーによるデータの処理に長い時間がかかる可能性があります。これはつまり、セッションのタイムアウトを回避するために、コンシューマーがポーリンググループの次の反復に間に合わないということを意味します。
  注: session.timeout.ms に高い値を設定すると、誤って再調整が行われる可能性が下がります。ただし、実際の障害の検出には時間がかかる場合があります。このパラメータは heartbeat.interval.ms に関連しています。heartbeat.interval.ms パラメータは、KafkaConsumer poll() メソッドがハートビートをグループコーディネーターに送信する頻度を制御します。ただし、session.timeout.ms パラメータは、コンシューマーがハートビートを送信せずに過ごせる時間を制御します。
  たとえば、Apache Kafka 0.10.1 以降を実行していて、処理に時間がかかるレコードに対応しているとします。この場合は、新しいレコードのポーリング間のより長い遅延に対応できるように max.poll.interval.ms を調整します。
• コンシューマー設定の session.timeout.ms の値が、ブローカー設定の group.max.session.timeout.ms の値よりも低いことを確認してください。
• max.poll.interval.ms では、コンシューマーが追加のレコードを取得するまでのアイドル状態でいられる時間の上限を設定します。デフォルトでは、この値は 5 分に設定されています。この値が 5 分未満に設定されている場合、値を大きくすると再調整が発生する可能性が低くなります。また、max.poll.records を、max.poll.interval.ms と一緒に減らすことも検討できます。
• heartbeat.interval.ms は、Kafkaのグループ管理機能を使用している場合に、ハートビートがコンシューマーコーディネーターに送信されるまでの予想時間です。ハートビートは、コンシューマーのセッションをアクティブ状態に保つために使用されます。これにより、新しいコンシューマーがグループに参加したり、グループから脱退したりしたときに、再調整しやすくなります。この値は、session.timeout.ms よりも低い値に設定する必要があります。通常、この値を session.timeout.msの 3 分の 1 以下の値に設定することが必要です。heartbeat.interval.ms の値をかなり低くして、通常の再調整にかかる予想時間を制御することができます。
• コンシューマーグループのサブスクライブ済みトピックのいずれかのパーティションへの変更が生じるパーティション再割り当てを最近実行した場合は、コンシューマーグループが再調整される可能性があります。これは、関係するパーティションが移動または変更されるためです。この場合、グループコーディネーターや他の Kafka ブローカーを再起動しないでください。コンシューマーグループの再調整を停止する前に、パーティションの再割り当ての完了を待つ必要があります。トラフィックの少ない時間帯にパーティションの再割り当てを行うのがベストプラクティスです。

場合によっては、Amazon MSK ブローカーのログに次の情報が表示されることがあります。

[2023-03-01 01:23:45,678] INFO [GroupCoordinator 1]: Preparing to rebalance group amazon.msk.canary.group.broker-1 in state PreparingRebalance with old generation 382660 (__consumer_offsets-21) (reason: Adding new member consumer-amazon.msk.canary.group.broker-1-xxxx-xxxx-xxxx-xxxx-xxxx-xxxx with group instance id None) (kafka.coordinator.group .GroupCoordinator)

このメッセージは、amazon.msk.canary.group.broker-N が PreparingRebalance 状態にあることを示しています。

amazon.msk.canary.group.broker-N グループは内部コンシューマーグループで、クラスターの状態と診断メトリックスをチェックするために定期的に追加または削除されます。これらのグループのサイズはごくわずかで、削除することはできません。このメッセージは無視できます。

関連情報

コンシューマーグループが PreparingRebalance 状態から変化しない