Kafka について: コンシューマーソースコード分析とリバランスメカニズム

[[443148]]

1. はじめに

前回の記事では、Consumer が Consumer Group に参加する仕組みを分析しました。実際、前回の記事は非常にマクロなもので、主に ConsumerCoordinator が GroupCoordinator とどのように通信するかについて説明しました。待ってください、Lao Zhou、ConsumerCoordinator と GroupCoordinator とは何ですか?これら 2 つのコンポーネントは、それぞれ Consumer と Kafka Broker のコーディネーターです。簡単に言えば、これらはデザインパターンにおけるファサードパターンです。具体的な内容については前回の記事を参照して確認することができます。今日の記事では、主に、コンシューマーがコンシューマーグループに参加する方法についての前回の記事で説明したリバランスメカニズムについて説明します。実際、前回の記事では概要を説明しましたが、この記事ではリバランスのメカニズムの具体的な詳細についてさらに詳しく説明します。

ある程度の経験があるプログラマーであれば、リバランスの仕組みは面接の質問として使えると思いますが、それでもかなり難しいです。しかし、自分自身を過小評価する必要はありません。この記事に従うだけで、必ず達成できると信じています。

しかし、一部の読者にとっては難しいと感じるかもしれません。心配しないで。以下の Kafka のトポロジをご覧ください。構造は非常に明確です。 Kafka のトポロジを理解していない場合は、読み進めないことをお勧めします。まず Kafka のトポロジーを明確に理解するか、読み進める前に Lao Zhou の以前の記事を読んでください。効果はもっと良くなると思います。

この記事では、主に以下の点からリバランスのメカニズムについて説明します。

リバランスメカニズムとは何ですか?
リバランスメカニズムをいつ起動するか
グループステータスの変更
高齢の消費者クライアントに関する問題
リバランスメカニズムの原理
ブローカー側のリバランスシナリオ

2. リバランスメカニズムとは何ですか?

リバランスは、本質的には、コンシューマーグループ内のすべてのコンシューマーが、サブスクライブされたトピックの各パーティションを割り当てるために合意に達する方法を指定するプロトコルです。

新しいメンバーがクラスターに参加したり、一部のトピックにパーティションが追加されたりすると、コンシューマーは消費をどのように再分配するのでしょうか?これにはリバランスの概念が含まれます。 Kafka の再バランス調整メカニズムとは何かを説明します。

この図から、消費者グループモデルのいくつかの概念がわかります。

同じコンシューマーグループでは、パーティションは 1 つのコンシューマーによってのみサブスクライブおよび消費できますが、コンシューマーは複数のパーティションをサブスクライブできます。つまり、各メッセージは同じコンシューマーグループ内の 1 つのコンシューマーによってのみ消費され、繰り返し消費されないことが保証されます。
パーティションは、異なるコンシューマーグループによってサブスクライブできます。ここでは特別なケースがあります。各コンシューマーグループにコンシューマーが 1 つしかない場合、パーティションはすべてのコンシューマーにブロードキャストされ、ブロードキャストモードの消費が実現されます。

上記のコンシューマグループモデルを実装するには、トピックに新しいパーティションが追加されたり、コンシューマグループに新しいメンバーが参加したりするなど、外部環境が変化したときに上記のモデルを維持するための動的な調整を実装する必要があります。この作業は、Kafka の再バランス調整メカニズムによって処理されます。

図からわかるように、Kafka の再バランスは外部トリガーによって発生します。 Kafka の再バランス調整をトリガーする機会を見てみましょう。

3. リバランスメカニズムをいつ起動するか

新たな消費者が消費者グループに加わりました
コンシューマーはオフラインです。消費者は必ずしもオフラインである必要はありません。たとえば、GC またはネットワークの遅延が長時間続いたために Consumer が GroupCoordinator に HeartbeatRequest を長時間送信しなかった場合、GroupCoordinator は Consumer をオフラインと見なします。
コンシューマーが積極的にコンシューマーグループを退出します (LeaveGroupRequest リクエストを送信します)。たとえば、クライアントは unsubscribe() メソッドを呼び出して、特定のトピックへのサブスクリプションをキャンセルします。
コンシューマーがタイムアウトになり、指定された時間内にオフセットを送信しませんでした。
コンシューマーグループに対応する GroupCoordinator ノードが変更されました。
コンシューマーグループによってサブスクライブされたトピック、またはトピックのパーティション数が変更されます。

4. グループステータスの変更

4.1 消費者側

Consumer 側のファサード ConsumerCoordinator は、AbstractCoordinator 抽象クラスを継承します。コーディネーター AbstractCoordinator の内部クラス MemberState では、コーディネーターの 4 つの状態 (未登録、再割り当て後に応答を受信せず、再割り当て後に応答を受信したが割り当てを受信せず、安定状態) を確認できます。

上記の消費者側の 4 つの状態の遷移を次の図に示します。

4.2 サーバー

Kafka サーバーの GroupCoordinator には、Empty、PreparingRebalance、CompletingRebalance、Stable、Dead の 5 つの状態があります。状態遷移は次の図に示されています。

コンシューマグループは、最初は空です。
リバランスが有効になると、メンバーが参加するのを待つ PreparingRebalance 状態になります。
次にCompletingRebalanceに変更し、割り当て計画を待ちます。
最後に、フローは安定版に転送され、リバランスが完了します。
メンバーが変更されると、コンシューマーグループの状態は Stable から PreparingRebalance に変わります。
- 現時点では、既存のメンバー全員がグループに参加するには再度申請する必要があります。
- すべてのグループメンバーがグループから退出すると、コンシューマーグループのステータスは空になります。
コンシューマグループが空の状態の場合、Kafka は期限切れのオフセットを定期的に自動的に削除します。

5. 旧バージョンの消費者クライアントの問題

ConsumerCoordinator と GroupCoordinator の概念は、Kafka バージョン 0.9.0 以降のコンシューマークライアント向けです。当面は、Kafka バージョン 0.9.0 より前のコンシューマークライアントを旧バージョンのコンシューマークライアントと呼びます。古いバージョンのコンシューマークライアントでは、これらの機能を実装するために Zookeeper の Watcher を使用します。

各消費者グループ/consumers/はZookeeperで管理されています/ids パス。このパスの下では、このコンシューマグループに属するコンシューマの一意の識別子 consumerldString を記録するために一時ノードが使用されます。 consumerldString は、コンシューマーが起動されたときに作成されます。コンシューマーの一意の識別子は、consumer.id + ホスト名 + タイムスタンプ + UUID の部分情報で構成されます。ここで、consumer.id はコンシューマークライアントの古いバージョンの構成であり、クライアントの新しいバージョンの client.id に相当します。たとえば、コンシューマーの一意の識別子が consumerld_localhost-1510734527562-64b377f5 の場合、consumerld は指定された consumer.id、localhost はコンピューターのホスト名、1510734527562 はタイムスタンプ、64b377f5 は UUID 情報の一部を表します。

次の図は消費者に関するものです。 /idsと同じレベルには、ownersとoffsetsという2つのノードがあります。

/consumers//ownersパスはパーティションとコンシューマー間の対応を記録します
/consumers//offsetsパスは、パーティション内のこのコンシューマグループの対応する消費オフセットを記録します。

各ブローカー、トピック、パーティションは Zookeeper 内のパスにも対応しています。

/brokers/ids/ には、このブローカーに割り当てられたホスト、ポート、およびトピックパーティションのリストが記録されます。
/brokers/topics/ には、各パーティションのリーダーレプリカや ISR セットなどの情報が記録されます。
/brokers/topics//partitions//state には、現在のリーダーレプリカやリーダーエポックなどの情報が記録されます。

各消費者は/消費者/にいます/ids および /brokers/ids パスにリスナーを登録します。 /消費者/ /ids パスの下のサブノードが変更されると、コンシューマーグループ内のコンシューマーが変更されたことを意味します。 /brokers/ids パスの下のサブノードが変更された場合、ブローカーが増加または減少したことを意味します。このように、Zookeeper が提供する Watcher を通じて、各コンシューマーはコンシューマーグループと Kafka クラスターのステータスを監視できます。

このようにして、各コンシューマーは Zookeeper の関連パスを個別に監視します。再バランス操作がトリガーされると、コンシューマーグループ内のすべてのコンシューマーが同時に再バランス操作を実行しますが、コンシューマーは互いの操作の結果を認識できないため、Kafka が誤った状態で動作する可能性があります。同時に、Zookeeper クラスターに大きく依存するこのアプローチには、さらに 2 つの深刻な問題があります。

群れ効果: いわゆる群れ効果は、Zookeeper の監視対象ノードの変更を指します。大量の Watcher 通知がクライアントに送信され、通知期間中に他の操作が遅延し、デッドロックのような状況が発生する可能性もあります。
スプリットブレイン: コンシューマーがリバランス操作を実行すると、各コンシューマーは Zookeeper と通信して、コンシューマーまたはブローカーの変更を判断します。 Zookeeper 自体の特性により、各コンシューマーが同時に取得した状態が矛盾する可能性があり、異常な問題が発生する可能性があります。

6. リバランスメカニズムの原則

Kafka バージョン 0.9.0 以降のコンシューマークライアントは、すべてのコンシューマーグループを複数のサブセットに分割するように再設計されました。コンシューマグループの各サブセットは、それを管理するサーバー上の GroupCoordinator に対応します。 GroupCoordinator は、コンシューマーグループを管理するために使用される Kafka サーバー内のコンポーネントです。コンシューマークライアントの ConsumerCoordinator コンポーネントは、GroupCoordinator との対話を担当します。

完全な再バランス調整プロセスには、コンシューマーとコーディネーターの共同の取り組みが必要です。
消費者側の再調整手順
- グループに参加: JoinGroupリクエストに対応
- リーダーコンシューマー割り当て計画を待機中: SyncGroup 要求に対応
メンバーがグループに参加すると、コンシューマーはコーディネーターに JoinGroup リクエストを送信します。
各コンシューマーはサブスクライブしているトピックを報告します
コーディネーターは、すべての JoinGroup リクエストを収集した後、これらのメンバーの中からコンシューマーグループのリーダーとして機能するリーダーを選択します。
- 通常、最初にJoinGroupリクエストを送信した人が自動的にリーダーになります。
リーダーコンシューマーのタスクは、すべてのメンバーからトピックを収集し、その情報に基づいて特定のパーティションコンシューマー割り当て計画を策定することです。
リーダーを選択した後、コーディネーターはすべてのトピック情報を JoinGroup 応答にカプセル化し、リーダーに送信します。
リーダーコンシューマーは、統一された割り当て計画を作成し、SyncGroup 要求を入力します。
リーダーコンシューマーは SyncGroup をコーディネーターに送信し、割り当て計画をコーディネーターに送信します。
他のメンバーもSyncGroupリクエストを発行します
コーディネーターは、SyncGroup Response の形式でソリューションをすべてのメンバーに送信します。