Zookeeper を廃止した後、Kafka はトピックとコンシューマーグループをどのように保存しますか?

筆者の会社で現在使用している Kafka のバージョンは 2.2.1 であるため、Kafka カーネルに関する現在の研究は主にこのバージョンに基づいています。もちろん、このコラムでは引き続き Kafka3.0 に焦点を当てていきます。

Kafka を使用していたとき、クライアントは Zookeeper に依存せずにメッセージを送信および消費できることが分かりました。ご存知のとおり、Kafka の初期の頃は、すべてのメタ情報 (トピック、コンシューマーグループ、クラスター) およびその他の情報は Zookeeper に保存されていました。元のメッセージ送信クライアントとメッセージ消費クライアントは Zookeeper に依存する必要がありました。

暖かいリマインダー: Kafka は徐々に動物園の飼育係から脱却し始めています。 Kafka 2.8 より前では、メッセージ送信者とメッセージコンシューマーは de-zookeeper でした。バージョン 2.8 以降では、ブローカーは de-zookeeper もサポートします。

では、Kafka 2.2.1 では、トピックルーティング情報とコンシューマーグループ情報はどこに保存されるのでしょうか?メッセージの送信者と消費者はそれをどのように認識するのでしょうか?

注意: Kafka について基本的な知識がある場合は、少し立ち止まって考えてみることをお勧めします。

1. トピックメタデータはZookeeperに保存されます

Kafka ブローカーに接続された Zookeeper クラスターに入ると、クラスター内のすべてのトピック情報が /{namespace}/brokers/topics ノード下に存在することが簡単にわかります。次の図に示すように、特定のトピックを展開します。

トピックに関するメタ情報には、主に以下の情報が含まれます。

パーティションの数各特定のトピックの下にパーティションノードがあり、このノードの下の各子ノードはパーティションを表します。
パーティションステータス情報各パーティションのステータスはリーフノード /{namespace}/brokers/topics/{topicName}/parttions/{partNO}/state で表され、保存される内容は次のとおりです。

controller_epoch コントローラーの現在の選択バージョン。

リーダーこのパーティションのリーダーが配置されているブローカーノード ID。

version 現在のストレージ形式のバージョン。デフォルトは 1 です。

leader_epoch パーティションリーダーの選出バージョン。

isr パーティションに設定された ISR。

トピックのルーティング情報は Zookeeper に保存されているのに、クライアントがトピックのルーティング情報を取得するためにブローカーアドレスのみを必要とするのはなぜですか?

1.1 トピックルーティングアドレス指定

Kafka 2.1 では、ApiKeys.METADATA リクエストを送信することによってルーティング情報が見つかります。このリクエストの応答ロジックはブローカーで定義されます。では、クライアントはブローカーをどのようにルーティングするのでしょうか。また、ブローカーのルーティング情報はどこから来るのでしょうか。

メッセージ送信者は、ブローカーメカニズムを見つけるために初めて METADATA を送信します。最初に送信されたリクエストでは、KafkaProducer の bootstrap.servers に設定されたブローカーリストから現在最もアイドル状態のブローカーが選択され、その後すべてのブローカーを検知できます。

メッセージコンシューマは、ブローカーメカニズムを見つけるために METADATA を送信します。これは、現在のコンシューマグループのグループ調整が配置されているブローカーに送信されます。

KafkaApis の handleTopicMetadataRequest メソッドによれば、コアメソッドに入る前にいくつかの ACL チェックが実行されます。

要点:

MetadataCache からトピックのルーティング情報を取得します。
指定されたトピックのルーティング情報が MetadataCache に存在せず、ブローカーがトピックの自動作成 (auto.create.topics.enable) を許可している場合 (デフォルトは true)、トピックの情報が自動的に作成され、ZooKeeper に書き込まれます。具体的な操作は以下のとおりです。

/brokers/topics ノードの下にサブノードを作成します。サブノード名はトピックの名前です。

現在の Kafka パーティションのラック情報、パーティション数、レプリカ数、ブローカーノード数に応じて割り当てが行われます。主な目的は、プライマリパーティションを同じラックに配置することを避け、トピックのノード情報に保存することです。たとえば、{"version":1,"partitions":{"4":[2,0,1],"5":[0,1,2],"1":[2,1,0],"0":[1,0,2],"2":[0,2,1],"3":[1,2,0]}} です。ここで、キーはパーティション名で、値はレプリカが配置されている brokerId です。最初のものは優先リーダーです。トピックに保存される値は静的データであり、選択をトリガーします。選挙アルゴリズムはこの割り当てを参照します。

また、コントローラーは、トピック情報の変更を監視するために registerPartitionModificationsHandlers メソッドを登録して呼び出し、それによって後続のプロセスをトリガーし、パーティションの実際の作成 (各パーティションのリーダー選出など) を開始します。

ヒント: Kafka が自動トピック作成を有効にすると、パーティションの数は Kafka ブローカーの num.partitions パラメータから取得され、デフォルトで 1 に設定されます。また、レプリケーション係数は default.replication.factor パラメータによって決まり、デフォルトで 1 に設定されます。

1.2 ルーティング情報同期メカニズム

MetadataCache、メタ情報キャッシュ、データはどこから来るのでしょうか? MetadataCache 内のルーティング情報の更新呼び出しチェーンを次の図に示します。

Kafka の KafkaController (以下、コントローラーと呼びます) は、まず /brokers/topics/{topicName} ノードのコンテンツの変更をリッスンします。新しいトピックが作成されるか、トピック情報が変更されると、トピック変更イベントがトリガーされます。このとき、TopicChange の process メソッドが呼び出され、最終的に updatePartitionReplicaAssignment が呼び出されます。つまり、トピックの情報が変更されると、コントローラーはすべてのブローカーノードに ApiKeys.UPDATE_METADATA を送信します。リクエストを受信すると、各ブローカーはメッセージ送信者がトピックルーティング情報を見つけられるように、各ブローカー内のメモリキャッシュを更新します。

つまり、Kafka 2.2 では、トピックのメタデータは Zookeeper に保存されます。同時に、Kafka コントローラーは Zookeeper 内の関連ノードを監視して情報の変更を認識し、RPC を介してクラスター内のすべてのブローカーにルーティング情報を送信します。したがって、各ブローカーは同じルーティング情報をメモリに保存します。

Kafka バージョン 2.8 では、Zookeeper の廃止を試み始めました。

検討すべき質問: 各ブローカーはなぜ Zookeeper をリッスンしてトピックの変更を感知し、ローカルメモリを更新しないのでしょうか?話し合いのためにメッセージを残したり、dingwpmz にプライベートメッセージを送信してコミュニケーションを取ることもできます。

2. コンシューマグループはロケーショントピックに保存されます

以前のバージョンでは、Kafka コンシューマーグループを起動するには、Zookeeper クラスターのアドレスを指定する必要がありました。これは、以前のバージョンでは、コンシューマーグループのメタデータが Zookeeper に保存され、特定のパスが /consumers であったためです。ただし、以降のバージョンでは、コンシューマー側を起動するときに Zookeeper を指定する必要はなく、ブローカーのアドレスリストを指定するだけで済みます。では、現時点で消費者グループの情報はどこに保存されているのでしょうか?

Kafka のトラブルシューティングに関する以前の記事では、コンシューマーグループメタデータマネージャー GroupMetadataManager を保持するコンシューマーグループコーディネーターがよく登場します。関連するコードのスクリーンショットは次のとおりです。

GroupMetadataManager オブジェクトは、キーがコンシューマグループの名前、値が GroupMetadata オブジェクトである Map 構造のキャッシュを保持します。GroupMetadata オブジェクトは、コンシューマグループの状態、コンシューマグループのメンバーリスト、および場所の情報を記録します。

メモリの特性: 効率的なアクセスですが、ブローカープロセスの終了時に失われます。明らかに、コンシューマグループをメモリに保存することはできませんが、Zookeeper では保存できません。では、コンシューマーグループの定義情報はどこに保存されるのでしょうか?

2.1 消費者コンポーネント情報の保存

コンシューマグループの定義情報は、システムトピック __consumer_offsets に保存されます。何？このトピックは消費者のオフセットを保存するために使用されませんか?

__consumer_offsets はコンシューマグループの位置情報を保存するだけでなく、コンシューマグループのメタデータも保存することがわかります。具体的なコードエントリは、GroupMetadataManager#storeGroup です。いくつかのコードのスクリーンショットを以下に示します。

つまり、消費グループのメタデータはメッセージとして __consumer_offsets に書き込まれます。消費グループメタデータに格納される値は、GroupMetadataManager の groupMetadataValue メソッドによって定義されます。具体的なコードは次のとおりです。

Kafka は進化を続けており、ストレージ形式は何度も変更されてきました。対応するバージョンは次のとおりです。

V0: Kafka バージョン 0.10 未満
V1: 0.10 より大きく、バージョン 2.1 以下。
V2: バージョン 2.2 以降

コンシューマーコンポーネント情報ストレージの形式は Json であり、具体的なストレージコンテンツは次のとおりです。

protocol_type プロトコルバージョンは、AbstractCoordinator の抽象メソッド protocolType() から取得され、コンシューマグループは consumer として固定されます。
世代コンシューマーグループメタデータのバージョン番号。この値は、コンシューマーグループの再バランスが発生するたびに 1 ずつ増加します。
プロトコルプロトコルコンテンツ。コンシューマグループのキューロードアルゴリズムが格納されます。これは、コンシューマーを構築するときに、partition.assignment.strategy パラメータを介して渡すことができます。複数の戦略を渡すこともできます。コンシューマグループの特定のロードアルゴリズムにより、キューロードに対して各コンシューマがサポートするプロトコルが選択されます。デフォルトのロードアルゴリズムは RangeAssignor です。
リーダー現在のコンシューマーグループのリーダー。通常はコンシューマーグループに参加する最初のコンシューマーです。
current_state_timestamp 最新の状態変更のタイムスタンプ。この値はバージョン V2 から導入されました。
メンバーコンシューマーグループのメンバー情報。各メンバーに保存される情報は次のとおりです。
member_id メンバー ID、クライアント ID (clientId) + uuid。

client_id クライアントID。

client_host クライアントの IP アドレス。

rebalance_timeout 再バランス時間。デフォルトは 300000、5 分です。

session_timeout セッションタイムアウト。デフォルト値は 10 秒です。

サブスクリプションメタデータは、AbstractCoordinator の抽象メソッド metadata() から取得されます。コンシューマグループの実装クラスは ConsumerCoordinator であり、主にロードアルゴリズムを走査します。各ロードアルゴリズムは、サブスクリプション情報に基づいてメタデータを計算します。

割り当て

各コンシューマーのキュー負荷。

ヒント: GroupMetadataManager の storeGroup メソッドは、コンシューマーグループが再バランス調整されるとき、具体的には再バランス調整の 2 番目のフェーズ (SYNC_GROUP) 中および再バランス調整が完了したときに呼び出されます。