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) 中および再バランス調整が完了したときに呼び出されます。

2.2 メッセージコンポーネント情報の読み込み

コンシューマー グループのメタデータは、__consumer_offsets トピックに保存されます。トピックからメモリにロードされるのはいつですか?

__consumer_offsets のパーティションでリーダー選出が発生すると、対応するパーティションのデータがメモリにロードされます。具体的な処理エントリは、KafkaApis の handleLeaderAndIsrRequest メソッドです。単純な呼び出しチェーンを次の図に示します。

3. まとめ

この記事では主に、Kafka トピックとコンシューマ グループの永続化メカニズムについて説明します。 Kafka 2.8 以降、公式では Zookeeper への依存を徐々に排除してきました。では、Kafka 3.x 以降ではコンシューマー グループとトピックの情報はどのように保存されるのでしょうか?