データ解放：Apache Kafkaのネイティブクラスターミラーリング

Data liberation: Apache Kafkaのネイティブクラスタミラーリング

Apache Kafkaは、クラスタ内でのデータ移動に優れています。リーダーはフォロワーにレプリケートし、コンシューマーはどのレプリカからでもプルでき、全体の仕組みは最小限の運用オーバーヘッドで動作します。クラスタ間でデータを移動することは、これほど簡単ではありませんでした。

組織が複数のKafkaクラスタを実行するのには、地理的分散、コンプライアンス境界、チームの分離、バージョンの分離など、正当な理由があります。しかし、データがクラスタに着地すると、忠実なコピーを別のクラスタに取得するには、常に外部ツール、慎重な調整、そして運用上の予期せぬ事態に対する健全な許容力が必要でした。

KIP-1279はこの状況を変えます。クラスタミラーリングは、クラスタ間レプリケーションをKafkaブローカーに直接組み込みます。外部プロセスなし、オフセット変換テーブルなし、再圧縮なし。宛先ブローカーは、フォロワーが既に使っているのと同じフェッチプロトコルを使用して、ソースクラスタからコミットされたレコードをフェッチし、バイト単位でローカルパーティションログに追加します。結果として、オフセット、圧縮、コンシューマーグループの状態を保持するミラーが作成され、フェイルオーバーがミラーを停止してクライアントをリダイレクトするのと同じくらい簡単になります。

この記事では、クラスタミラーリングの背後にある高レベルアーキテクチャ、ミラーパーティションを管理するステートマシン、そしてそれらすべてをまとめる一貫性保証について説明します。その後、ディザスタリカバリとクラスタ移行の2つの実践的なシナリオを説明します。

クラスタ間レプリケーションの新しいアプローチ

MirrorMaker 2 (MM2) は、Kafka 2.4以来、クラスタ間レプリケーションの標準ツールとして機能しており、ソースクラスタから消費して宛先クラスタにプロデュースするKafka Connectワーカーのセットとして実行されます。クラスタミラーリングは、レプリケーションをブローカーに直接組み込むことで、根本的に異なるアプローチを取ります。

• ゼロ外部インフラストラクチャ: ミラーフェッチャー スレッドは、ブローカープロセス自体内で実行されます。プロビジョニング、監視、スケーリングするConnectワーカーはありません。単一のCLIコマンド (kafka-cluster-mirrors.sh --create) でミラーが確立され、別のコマンド (--start) でトピックのレプリケーションが開始されます。ライフサイクルの全体は、トピックやコンシューマーグループに使用されるのと同じAdmin APIを通じて管理されます。
• バイト単位の転送: 圧縮されたバッチは、解凍または再圧縮されることなく、生のバイトとしてレプリケートされます。gzip、snappy、lz4、またはzstdバッチは、元の形式で宛先に到着します。これにより、解凍/再圧縮の往復処理のCPUオーバーヘッドが排除され、プロデューサーの元の圧縮選択が維持されます。
• 正確なオフセット保持: 宛先ログは、トピックコンパクションによって残されたギャップを含め、ソースと同じオフセットを維持します。コンシューマーグループは、オフセット変換なしでフェイルオーバーします。ソースでのコミット済みオフセットは、宛先でのコミット済みオフセットです。
• ワンコマンドフェイルオーバー: ミラーを停止する (--stop) と、パーティションは決定論的なシーケンスを通過します。フェッチャーが削除され、最後のミラーエポックが永続化され、リーダーエポックが引き上げられ、保留中のトランザクションが中止され、新しい制御レコードがすべてのプロデューサー状態を期限切れにします。パーティションは宛先で書き込み可能になります。外部調整やオフセットクエリは不要です。
• アンクリーンリーダー選出のサポート: ソースクラスタでアンクリーンリーダー選出 (ULE) が発生した場合、宛先はリカバリ状態に入ります。レプリケーションを再開する前に、ISRメンバーだけでなく、すべての割り当てられたレプリカが切り捨てられたオフセットに収束するのを待ちます。これにより、ソースが不完全なログを持つリーダーを選出した場合でも、クラスタ間のログの一貫性が保証されます。

次の表は、主な違いをまとめたものです。

クラスタミラーリングにより、宛先ブローカーはクラスタ間レプリケーションのアクティブな参加者になります。各ブローカーは、標準のフェッチプロトコルを使用してソースクラスタから直接データをフェッチし、生のレコードバッチをローカルパーティションログに追加します。ソースと宛先のパーティションは同じトピックIDを共有します。

データレプリケーションを超えて、ブローカーはメタデータ検出、設定同期、グループオフセット同期、ACL伝播も処理します。帯域幅制御は両側で機能します。宛先ブローカーは、設定可能なレプリケーションレート制限を強制します。ソース側では、ミラーフェッチトラフィックは標準のコンシューマーリクエストとして表示されるため、既存のクライアントクォータメカニズムは変更なしで適用されます。

アーキテクチャ

各宛先ブローカー内では、3つの主要なコンポーネントが連携しています。図1は、それらが互いに、およびソースクラスタにどのように接続されているかを示しています。

［画像: Cluster Mirroring architecture inside a destination broker.］

図1: 宛先ブローカー内のクラスタミラーリングアーキテクチャ。

MirrorMetadataManager

MimorMetadataManager (MMM) はオーケストレーターです。すべてのブローカーで実行され、MetadataPublisherインターフェースを実装して、KRaftメタデータログの変更に反応します。コントローラーがMirrorTopicStateChangeRecordを書き込むと、影響を受けるパーティションのリーダー上のMMMが、特定の操作（作成、開始、停止、一時停止、再開、リカバリ、削除）をトリガーする対応する状態遷移を駆動します。

MMMはソースクラスタへのAdminクライアント接続も維持します。デフォルトで60秒ごとに、ソースメタデータをリフレッシュします。設定されたinclude/excludeパターンに一致する新しいトピックの検出、トピック設定の同期、コンシューマーグループオフセットの取得、およびソースクラスタIDが変更されていないことの検証を行います。最後のチェックは、誰かが誤ってミラーを別のクラスタを指すように再設定した場合のサイレントデータ破損を防ぎます。

ClusterMirrorCoordinator

ClusterMirrorCoordinator (CMC) は状態永続化を処理します。グループコーディネーターやトランザクションコーディネーターが使用するのと同じコーディネーターパターンに従い、__mirror_state という内部コンパクト化トピックのシャードを管理します（デフォルト：コンパクションクリーンアップポリシー、50パーティション、レプリケーションファクター3）。各ミラーパーティションの状態は、このトピック内のキーバリューレコードとして保存され、リーダーエポックと状態エポックのフェンシングを通じて楽観的同時実行制御が行われます。

MirrorFetcherThread

MirrorFetcherThread (MFT) が重い作業を行います。KafkaのAbstractFetcherThread（クラスタ内レプリケーションに使用されるのと同じ基底クラス）を拡張し、ソースからレコードをフェッチしてローカルログに追加します。各スレッドは、ミラーごとの認証情報を持つ専用のNetworkClientを維持し、ミラー間でSASL/SSLコンテキストを分離します。フェッチャーマネージャーは、3次元識別子（フェッチャーID、ソースブローカーエンドポイント、ミラー名）でスレッドをキーイングし、きめ細かなロードバランシングとソースでのリーダー変更への迅速な応答を可能にします。

ミラーパーティションのライフサイクル

ミラーパーティションは、明確に定義された一連の状態を進行します。図2は、完全なステートマシンを示しています。

［画像: Mirror partition state machine. Any state can transition to FAILED on error.］

図2: ミラーパーティションのステートマシン。エラーが発生すると、どの状態からでもFAILEDに遷移できます。

• LOG_ALIGNMENT: 最初のステップは、ローカルログをソースに合わせることです。最初のミラーリングまたはサポートされていないソースのためにLME（Last Mirror Epoch）が見つからない場合、ブローカーはゼロに切り捨てて最初からミラーリングします。それ以外の場合は、デルタのみをミラーリングします。LMEレコードは、前のセッションからの最後のミラーオフセットとエポックを格納します。ブローカーは、最後のミラーオフセットまでのオフセットと、最後のミラーエポックまでのエポックでのみレコードを切り捨てます。これにより、あらゆる不一致が解決されます。
• EPOCH_FENCING: ブローカーはコントローラーにBumpLeaderEpochsリクエストを送信し、ローカルリーダーエポックを10ずつ増やします。再バンプしきい値は3です。この保証がないと、宛先のコンシューマーはソースからのコミット済みエポックで初期化される可能性があり、それはローカルエポックを超えているため、リーダーを古いと拒否し、停止します。
• MIRRORING: フェッチャースレッドはソースからのレコードのプルを開始します。圧縮されたバッチは、再圧縮なしでローカルログに直接追加されます。ローカルフォロワーがデータをレプリケートするにつれて、高ウォーターマークが進みます。グループオフセットはソースから定期的に同期され、宛先の有効なオフセット範囲にクランプされます。
• ULE_RECOVERY: ミラーリング開始後、ソースクラスタでのULE（Unclean Leader Election）のみがログの不一致を引き起こす可能性があります。コミットされたレコードのみがミラーリングされるためです。ULEサポートが有効で、宛先リーダーがフェッチ中にログの不一致を検出すると、パーティションはMIRRORINGからこの状態に遷移します。ミラーフェッチャーは削除され、システムはISRメンバーだけでなく、すべてのレプリカが切り捨てられたログエンドオフセットに収束するのを待ちます。すべてのレプリカが追いつくと、パーティションはMIRRORINGに戻り、新しいフェッチャーが作成されます。
• PAUSING / PAUSED: オペレーターは、進捗を失うことなくミラーリングを一時停止できます。フェッチャースレッドは破棄されますが、パーティションは読み取り専用のままです。再開すると、新しいフェッチャーで直接MIRRORINGに戻り、再アライメントは行われません。
• STOPPING / STOPPED: これはフェイルオーバーパスです。ブローカーはフェッチャースレッドを削除し、LMEを将来のフェイルバックのために記録し、ローカルエポックを引き上げ、保留中のトランザクションに対してABORTマーカーを追加し、MIRROR_PID_RESET制御レコードを書き込んで、すべてのプロデューサー状態を期限切れにします。その後、パーティションは書き込み可能になります。停止したミラーを再起動すると、トピックが書き込み可能であった間にローカル書き込みを蓄積した可能性があるため、LOG_ALIGNMENTを通じて遷移します。
• FAILED: エラーは、設定可能な最大試行回数まで、指数関数的なバックオフ（ジッター付き）で自動再試行をトリガーします。リトライ不可能なエラーは、手動でのリカバリが必要です（例：ソースクラスタIDの変更、ソーストピックの削除）。

データの一貫性

宛先クラスタのログをソースと一貫して保つための2つのメカニズムがあります。ログ収束はエポックアライメントと切り捨てを処理し、トランザクション安全性はフェイルオーバー後にコミットされていないレコードが漏洩しないことを保証します。

ログ収束

宛先リーダーエポックは、常にソースリーダーエポック以上（DLE >= SLE）である必要があります。この不変条件がないと、宛先のコンシューマーはソースからのコミット済みエポックで初期化される可能性があり、それはローカルエポックを超えているため、フェッチを拒否し、停止します。システムは、リアクティブバンプ（フェッチされたバッチエポックがローカルエポックに近づいたとき）、プロアクティブバンプ（ギャップが3未満に縮小したとき）、および定期バンプ（ソースメタデータ同期中）を通じてギャップを維持します。各バンプはエポックを10ずつ増やします。

ミラーリングが開始または再開されると、ローカルログとソースログは不一致になっている可能性があります。2フェーズ切り捨てプロトコルがこれを解決します。

• 最初のフェーズは、リーダーエポック履歴をアラインします。LMEレコードは、前のミラーリングセッションからの最後のミラーオフセットとエポックを格納します。LOG_ALIGNMENT中、ブローカーは最後のミラーオフセットまでのオフセットでのみレコードを切り捨てます。これにより、あらゆる不一致が解決されます。LMEがない場合、ミラーリングは最初から開始されます。
• 定常状態フェーズは、通常のフェッチ処理中に残りのエポック内のオフセットをアラインします。フェッチャーはソースクラスタのフェッチバージョンを動的に検出します。ソースがFetch v12+をサポートしている場合、フェッチレスポンス内の不一致エポック情報は、正確な不一致ポイントへのインライン切り捨てを可能にします。サポートしていない場合、フェッチャーはフェッチ時の切り捨てを無効にし、明示的なOffsetsForLeaderEpochリクエストにフォールバックして切り捨てポイントを見つけます。これは、ライブアップグレードやダウングレードを含む、古いクラスタと新しいクラスタの両方を処理します。

ミラーリング中に宛先がコミットされたデータのみをフェッチするため、LOG_ALIGNMENT後のあらゆる切り捨ては、ソースでのULEに起因する必要があります。これはフェッチバージョンに関係なく適用されます。mirror.unclean.leader.election.enable 設定により、オペレーターは明確な選択ができます。trueに設定されている場合、パーティションはULE_RECOVERYに遷移し、レプリケーションを再開する前にすべての割り当てられたレプリカが収束するのを待ち、ソースULEからのデータ損失を受け入れます。false（デフォルト）に設定されている場合、パーティションはFAILED状態に移行し、ミラーリングを完全に停止します。

トランザクション安全性

クラスタミラーリングはREAD_UNCOMMITTED分離でレプリケートします。これは、ソースがトランザクションの運命を決定する前に、コミットされていないトランザクションレコードが宛先に到着することを意味します。これは意図的です。トランザクション完了を待つと、データパスのレイテンシと複雑さが増加します。

停止遷移中、ブローカーは保留中のトランザクションに対してABORTマーカーを追加し、READ_COMMITTEDコンシューマーがフェイルオーバー後に保留中のトランザクションを見ないことを保証します。また、MirrorPidResetRecord 制御レコードを書き込み、ProducerStateManager のすべてのプロデューサー状態エントリを期限切れにします。これにより、宛先（書き込み可能になった）の新しいプロデューサーが競合なしで新しいプロデューサーIDを取得できるようになります。

PIDリセットメカニズムは、すべてのレプリケーショントポロジー（アクティブ-パッシブ（AからB）、フェイルバックチェーン（AからBからA）、ファンアウト（AからBおよびAからC）、マルチホップチェーン（AからBからC））で正しく伝播します。

ディザスタリカバリ

ディザスタリカバリは、クラスタミラーリングの最も自然なアプリケーションです。図3は、3つのフェーズを示しています。

［画像: Disaster recovery phases: normal operation, failover, and failback.］

図3: ディザスタリカバリのフェーズ：通常操作、フェイルオーバー、およびフェイルバック。

通常操作

クラスタAがすべての本番トラフィックを提供します。a-to-bという名前のミラーは、トピックをクラスタBに継続的にレプリケートします。

クラスタBで、ミラーを作成して開始します。

kafka-cluster-mirrors.sh --bootstrap-server B:9092 \
  --create --mirror a-to-b --mirror-config mirror.properties

--describe フラグを使用すると、いつでもレプリケーションの進捗を監視できます。

kafka-cluster-mirrors.sh --bootstrap-server B:9092 \
  --describe --mirror a-to-b

フェイルオーバー

クラスタAがダウンします。クラスタBでの単一コマンドでミラーリングが停止し、すべてのトピックが書き込み可能になります。

kafka-cluster-mirrors.sh --bootstrap-server B:9092 \
  --stop --mirror a-to-b

プロデューサーとコンシューマーは、ブートストラップサーバーをクラスタBに切り替えます。オフセットは同一であり、コンシューマーグループオフセットは同期されているため、アプリケーションは中断したところから再開します。オフセット変換なし。再処理なし。

フェイルバック

クラスタAが回復したら、オペレーターはリバースミラーを設定します。システムは、Aが以前これらのトピックのソースであったことを検出し、フェイルオーバー中に記録されたLMEを検索し、データセット全体を再レプリケートするのではなく、増分切り捨てを実行します。

クラスタAで、リバースミラーを設定します。

kafka-cluster-mirrors.sh --bootstrap-server A:9092 \
    --create --mirror b-to-a --mirror-config reverse.properties

レプリケーションが追いついたら、カットオーバーします。

kafka-cluster-mirrors.sh --bootstrap-server A:9092 \
  --stop --mirror b-to-a

リカバリポイント目標（RPO）は、障害発生時のレプリケーションラグに依存します。クラスタミラーリングは非同期であるため、Aで生成され、まだBにレプリケートされていないレコードは、予期しないフェイルオーバー中に失われます。ほとんどのDRユースケースでは、これは許容できるトレードオフです。同期クラスタ間レプリケーションはレイテンシペナルティを課し、プライマリクラスタのパフォーマンスを損なう可能性があります。KIP-1360（KIP-1360: Cluster Synchronous Mirroring）に注意してください。これはKIP-1279を同期モードに拡張する予定です。

クラスタ移行

クラスタミラーリングは、Kafka 4.0で導入されたクライアント/ブローカーのフォワード互換性を活用し、Kafka 2.1まで遡るソースブローカーをサポートします。これにより、従来の段階的なバージョンアップを完全に回避する実用的な移行パスが開かれます。図4は、3つのフェーズを示しています。

［画像: Cluster migration from an old ZK-based cluster to a new KRaft cluster.］

図4: 古いZooKeeperベースのクラスタから新しいKRaftクラスタへのクラスタ移行。

従来の移行パスでは、すべての間の中間メジャーバージョン（2.xから3.xから4.x）をアップグレードし、インプレースでZooKeeperからKRaftへのメタデータ移行を実行し、何も壊れないことを願う必要があります。クラスタミラーリングを使用すると、プロセスは次のようになります。

1. ターゲットKafkaバージョンを実行する新しいKRaftクラスタを起動します。
2. 古いクラスタから新しいクラスタへのミラーを作成します。
3. ミラーリングを開始します。宛先はトピックを自動的に検出し、一致するパーティション（同一のトピックIDを持つ）を作成し、設定を同期し、データのレプリケーションを開始します。
4. 古いクラスタでプロデューサーを停止します。
5. レプリケーションラグが0に達するまで監視します。
6. ミラーリングを停止します（新しいクラスタのトピックが書き込み可能になります）。
7. クライアントを新しいクラスタにリダイレクトします。
8. 古いクラスタを廃止します。

新しいクラスタで：

kafka-cluster-mirrors.sh --bootstrap-server new:9092 \
    --create --mirror old-to-new --mirror-config old-cluster.properties

進捗を監視します：

kafka-cluster-mirrors.sh --bootstrap-server new:9092 \
    --describe --mirror old-to-new

ラグがゼロに達したら、カットオーバーします。

kafka-cluster-mirrors.sh --bootstrap-server new:9092 \
    --stop --mirror old-to-new

中間バージョンアップなし。インプレースメタデータ移行なし。ZooKeeperからKRaftへの変換スクリプトなし。古いクラスタは、シャットダウンする準備ができるまで、邪魔されずに実行されます。

結論

クラスタミラーリングは、Apache Kafkaがクラスタ間データ移動を処理する方法における変化を表しています。KIP-1279は、レプリケーションをブローカーに組み込むことで、運用上の複雑さのクラス全体を排除し、データ忠実度を向上させます。圧縮バッチはそのまま通過し、オフセットはクラスタ間で同一に保たれ、コンシューマーフェイルオーバーには変換が不要になります。

ディザスタリカバリの場合、一連のコマンドがマルチステップのランブックを置き換えます。クラスタ移行の場合、ZooKeeperベースのデプロイメント（Kafka 2.1実行中）から最新のKRaftクラスタへのクリーンなアップグレードパスを提供し、中間バージョンホップはありません。将来の作業には、ゼロRPO要件のための同期ミラーリング、階層型ストレージ統合、およびディスクレス トピックのサポートが含まれます。

あなたのデータは自由になりたがっているので、リスクのためにクラスタ移行を延期していたり、ランブックの束と十分なメンテナンスウィンドウを伴うDRドリルを実行していたりする場合、クラスタミラーリングはより簡単なパスを提供しますので、試してみてください。ハッピーハッキング。

最終更新: 2026年10月18日