Kafka のパーティションリーダーを変更する方法

数日前、私のグループの友人が私に尋ねました: Kafka の優先度レプリカを変更するにはどうすればいいですか?パーティション内のレプリカの 1 つをリーダーとして指定する必要があります。

ここに画像の説明を挿入

需要分析

この問題は、私たちの生産環境では非常によく発生します。多くの場合、トピック内のパーティションのリーダーを変更する必要があります。

たとえば、topic1-0には3つのレプリカ[0,1,2]があります。「優先レプリカ」ルールに従って

その場合、レプリカ 0 がリーダーである必要があります。パーティション内のリーダーレプリカのみが読み取り/書き込みレプリカを提供することは誰もが知っています。

その他のコピーはバックアップとして使用されます。場合によっては、コピー0のパフォーマンスリソースが不足していたり、ネットワークが良好でなかったり、IOの負荷が高かったりするなど、

これは間違いなく、トピックの全体的な読み取りおよび書き込みパフォーマンスに大きな影響を与えます。このとき、リーダーとしてプレッシャーの少ないレプリカに切り替えることが非常に重要です。

優先レプリカ: パーティション内のAR(すべてのレプリカ)情報、最初のレプリカがリーダーとして選択される

リーダーメカニズム: パーティションには読み取りと書き込みを担当するリーダーが 1 つだけ存在し、他のレプリカはバックアップとしてのみ使用されます。

では、そのような要件をどのように達成するのでしょうか?

解決

原理を理解すれば、それに応じた解決策を考えることができます。パーティションの AR の最初の位置を、指定したコピーに置き換えるだけです。 AR = {0,1,2} ==> AR = {2,1,0}

この目標を達成するには、一般的に 2 つの解決策があります。以下で分析してみましょう。

解決策1: パーティションレプリカの再配布

これまでにパーティションレプリカの再配布に関する記事を多数書きました。パーティションレプリカの再配布、データ移行、レプリカの拡張と縮小について詳しく知りたい場合は、リンクされた記事をお読みください。ここで簡単にお話しします。

一般に、パーティションレプリカの再配布には主に 3 つのプロセスがあります。

推奨移行JSONファイルを生成する
移行JSONファイルを実行する
移行プロセスが完了したことを確認する

ここでは主にステップ2を見て、移行ファイルが一般的にどのようになっているかを確認します。

 {
 「バージョン」 : 1,
 「パーティション」 : [{
 「トピック」 : 「トピック1」 、
 「パーティション」 : 0,
 「レプリカ」 : [0,1,2]
 }]
 }

この移行 Json は、topic1 のパーティション 0 のレプリカが [0,1,2] に分散されることを意味し、つまり topic1 のパーティション 0 には最終的に {brokerId-0,brokerId-1,brokerId-2} に 3 つのレプリカが存在することになります。

以前書いたパーティションレプリカ再分配原則のソースコード分析を読んだことがあるなら、間違いなく知っているだろう。

以前の割り当て方法が何であっても、最終的なコピー割り当ては [0,1,2] になります。コピー数が多いものは削除され、コピー数が少ないものが追加されます。

それで、私たちのニーズを満たしたい場合は、このJSONファイル内の「replicas」: [0,1,2]を変更するだけでいいのでしょうか?

たとえば、これを "replicas": [2,1,0] に変更します。Json を変更した後、それを実行すると、再配布プロセスが正式に開始されます。移行が完了すると、リーダーが上記の最初の位置のレプリカ「2」になっていることがわかります。

長所と短所

利点: 要件が満たされ、リーダーが積極的に切り替えられる

デメリット: 操作が複雑で、エラーが発生しやすくなります。まず元のパーティション割り当てデータを取得し、次に Json ファイルを手動で変更する必要があります。ここでは間違いが起きやすく、その影響も比較的大きくなります。もちろん、これらは検証インターフェースを通じて制限できます。最も重要な点は、レプリカの再配布には現時点では 1 つのタスクしか存在しないことです。現在「レプリカ再配布」タスクがある場合、ここでは実行できません。「レプリカの再配布」は比較的「重い」操作であり、エラーはクラスターに大きな影響を与えます。

解決策2: AR注文を手動で変更する

まず、パーティションレプリカ割り当てデータは Zookeeper のノード brokers/topics/{topicName} に保存されていることがわかります。 Topic1 のノードデータの例を見てみましょう。

 {
 「バージョン」 : 2,
 「パーティション」 : {
 「2」 ：[3, 2, 1],
 「1」 ：[2, 1, 3],
 「4」 ：[2, 3, 1],
 「0」 : [1, 3, 2],
 「3」 ：[1, 2, 3]
 },
 「レプリカの追加」 : {},
 「レプリカを削除しています」 : {}
 }

データの説明: バージョン: バージョン情報。現在、「1」と「2」の 2 つのバージョンがあります。

removed_replicas: 削除する必要があるレプリカデータ。パーティションレプリカの再配布プロセス中に、データ移行がほぼ完了すると、余分なレプリカは削除されます。削除が成功すると、ここのデータは消去されます

adding_replicas: 追加する必要があるレプリカデータ。パーティションレプリカの再配布プロセス中に、新しく追加されたレプリカが追加されます。追加が完了すると、ここにあるデータはクリアされます。

パーティション: トピックのすべてのパーティションコピーの割り当て方法。上記は、合計 5 つのパーティションとそれに対応するコピー場所があることを示しています。

これを知った上で、優先レプリカを変更したい場合は、Zookeeper 内のノードデータを直接変更するだけでよいのでしょうか?たとえば、パーティション「1」のレプリカの場所を[2,1,3]に変更します。

これを変更した後でも、たとえばkafkaコマンドを実行して、優先レプリカを再選出する必要があります。

 sh bin/kafka-leader-election.sh --bootstrap-server xxxx:9090 --topic Topic1--election-type PREFERRED --partition 1

--election-type : PREFERRED これは優先コピーモードでの再選出を意味します

つまり、これら 2 つの手順を完了すると、優先コピーを変更するという目標は達成されたことになります...?

実際にはそうではありません。なぜなら、ここではZookeeperノードのデータのみが変更されるからです。

bin/kafka-leader-election.sh の再選出操作はコントローラーによって実行されます。コントローラーの役割とソースコードを十分に理解している場合

コントローラーは各トピックのパーティションレプリカ情報を JVM メモリに保存することがわかっています。次に、Zookeeper 内のノードを手動で変更します。これにより、コントローラーが独自のメモリを更新するようにトリガーされません。

つまり、kafka-leader-election.sh を実行しても、優先レプリカは変更を認識しないため、まったく変更されません。

この問題の解決も非常に簡単です。コントローラーにデータの変更を認識させるだけです。最も簡単な方法は、コントローラーにデータを再選択させて再ロードさせることです。

要約する

Zookeeper の「AR」順序を手動で変更する

コントローラーの再選出

パーティションレプリカの再選出操作を実行する（優先レプリカ戦略）

シンプルなコード

もちろん、上記の機能は LogiKM に統合する必要があります。簡単なコードは以下のとおりです

// ここで HashMap 型に変換します。Kafka ノード データが後で新しいデータ ノードを追加してデータが失われるのを防ぐために、型をカスタマイズしないでください。
ハッシュマップのパーティションマップ = zkConfig.get(ZkPathUtil.getBrokerTopicRoot(topicName), HashMap.class);
 JSONObject パーティションJson = (JSONObject)partitionMap.get( "パーティション" );
 JSONArray パーティション = (JSONArray)partitionJson.get(partition); 
 
 //コードの一部は省略されています
 
 //スイッチシーケンスの優先順位のコピー
整数 最初= パーティション.getInteger(0);
パーティションを設定します(0、ターゲットブローカー);
パーティションを設定します(インデックス、最初); 
 
クラスター内の Zookeeper インスタンスを取得します。
文字列 json = JSON.toJSONString(partitionMap); 
 
 zkUtils.updatePersistentPath(ZkPathUtil.getBrokerTopicRoot(topicName), json, null ); 
 
 //書き込みが成功したら、非同期優先コピー選択をトリガーします
新しいスレッド(()->{
    試す {
        // 1. 最初にコントローラーを再選出します (そうしないと、上記の変更は有効になりません) (TODO.. 最適化が必要 -> コントローラーの再選出が頻繁に行われると、クラスターのパフォーマンスに影響します)
        zkConfig.deletePath(ZkPathUtil.CONTROLLER_ROOT_NODE);
        // コントローラーが選出されるまで待つ
        スレッドをスリープ状態にします(1000);
        //2.次にレプリカの再選出を開始する
        preferredReplicalElectCommand.preferredReplicaElection(クラスター ID、トピック名、パーティション、 "" );
    } キャッチ (ConfigException | InterruptedException e) {
        LOGGER.error( "再選例外.e:{}" ,e);
        e.printStackTrace();
    } 
 
 }）。始める（）;