第9章 MirrorMaker 2 での AMQ Streams の使用

MirrorMaker 2 は、ソース Kafka クラスターからのメッセージを消費し、ターゲット Kafka クラスターに書き込みます。

MirrorMaker 2 は以下を使用します。

MirrorMaker 2 は、クラスター間のデータ転送を管理する コネクター である Kafka Connect フレームワークに基づいています。

MirrorMaker 2 を設定して、ソースクラスターとターゲットクラスターの接続の詳細を含む Kafka Connect デプロイメントを定義し、一連の MirrorMaker 2 コネクターを実行して接続を確立します。

MirrorMaker 2 は次のコネクターで設定されます。

ソースクラスターからターゲットクラスターへのデータの ミラーリング プロセスは非同期です。各 MirrorMaker 2 インスタンスは、1 つのソースクラスターから 1 つのターゲットクラスターにデータをミラーリングします。複数の MirrorMaker 2 インスタンスを使用して、任意の数のクラスター間でデータをミラーリングできます。

図9.1 2 つのクラスターにおけるレプリケーション

デフォルトでは、ソースクラスターの新規トピックのチェックは 10 分ごとに行われます。頻度は、refresh.topics.interval.seconds をソースコネクター設定に追加することで変更できます。

9.1.1.2. 双方向レプリケーション (active/active)

リンクのコピー

MirrorMaker 2 アーキテクチャーは、アクティブ/アクティブ クラスター設定での双方向レプリケーションをサポートします。

各クラスターは、source および remote トピックの概念を使用して、別のクラスターのデータをレプリケーションします。同じトピックが各クラスターに保存されるため、リモートトピックの名前は MirrorMaker 2 によってソースクラスターを表すように自動的に変更されます。元のクラスターの名前の先頭には、トピックの名前が追加されます。

図9.2 トピック名の変更

ソースクラスターにフラグを付けると、トピックはそのクラスターにレプリケーションされません。

remote トピックを介したレプリケーションの概念は、データの集約が必要なアーキテクチャーの設定に役立ちます。コンシューマーは、同じクラスター内でソースおよびリモートトピックにサブスクライブできます。これに個別の集約クラスターは必要ありません。

MirrorMaker 2 は、ソースクラスターとターゲットクラスター間のトピック設定の同期をサポートします。MirrorMaker 2 設定でソーストピックを指定します。MirrorMaker 2 はソーストピックを監視します。MirrorMaker 2 は、ソーストピックへの変更を検出し、リモートトピックに伝達します。変更には、欠けているトピックおよびパーティションの自動作成が含まれる場合があります。

MirrorMaker 2 は、内部トピックを使用してコンシューマーグループのオフセットを追跡します。

これらは MirrorMaker 2 によって内部的に使用されるため、これらのトピックと直接対話することはありません。

MirrorCheckpointConnector は、オフセット追跡用の チェックポイント を発行します。checkpoints トピックのオフセットは、設定によって事前に決定された間隔で追跡されます。両方のトピックは、フェイルオーバー時に正しいオフセットの位置からレプリケーションの完全復元を可能にします。

offset-syncs トピックの場所は、デフォルトで source クラスターです。offset-syncs.topic.location コネクター設定を使用して、これを target クラスターに変更することができます。トピックが含まれるクラスターへの読み取り/書き込みアクセスが必要です。ターゲットクラスターを offset-syncs トピックの場所として使用すると、ソースクラスターへの読み取りアクセス権しかない場合でも、MirrorMaker 2 を使用できるようになります。

__consumer_offsets トピックには、各コンシューマーグループのコミットされたオフセットに関する情報が保存されます。オフセットの同期は、ソースクラスターのコンシューマーグループのコンシューマーオフセットをターゲットクラスターのコンシューマーオフセットに定期的に転送します。

オフセットの同期は、特に active/passive 設定で便利です。アクティブなクラスターがダウンした場合、コンシューマーアプリケーションを passive (スタンバイ) クラスターに切り替え、最後に転送されたオフセットの位置からピックアップできます。

トピックオフセットの同期を使用するには、sync.group.offsets.enabled を checkpoint コネクター設定に追加し、プロパティーを true に設定して、同期を有効にします。同期はデフォルトで無効になっています。

ソースコネクターで IdentityReplicationPolicy を使用する場合は、チェックポイントコネクター設定でも設定する必要があります。これにより、ミラーリングされたコンシューマーオフセットが正しいトピックに適用されます。

コンシューマーオフセットは、ターゲットクラスターでアクティブではないコンシューマーグループに対してのみ同期されます。コンシューマーグループがターゲットクラスターにある場合、Synchronization を実行できず、UNKNOWN_MEMBER_ID エラーが返されます。

同期を有効にすると、ソースクラスターからオフセットの同期が定期的に行われます。この頻度は、sync.group.offsets.interval.seconds および emit.checkpoints.interval.seconds をチェックポイントコネクター設定に追加することで変更できます。これらのプロパティーは、コンシューマーグループのオフセットが同期される頻度 (秒単位) と、オフセットを追跡するためにチェックポイントが生成される頻度を指定します。両方のプロパティーのデフォルトは 60 秒です。refresh.groups.interval.seconds プロパティーを使用して、新規コンシューマーグループのチェック頻度を変更することもできます。デフォルトでは 10 分ごとに実行されます。

同期は時間ベースであるため、コンシューマーによって passive クラスターへ切り替えられると、一部のメッセージが重複する可能性があります。

MirrorHeartbeatConnector は heartbeat を発行して、クラスター間の接続を確認します。

内部 heartbeat トピックは、ソースクラスターからレプリケーションされます。ターゲットクラスターは、heartbeat トピックを使用して次のことを確認します。