9.2. MirrorMaker 2 コネクターの設定
Kafka クラスター間のデータの同期を調整する内部コネクターには、MirrorMaker 2 コネクター設定を使用します。
MirrorMaker 2 は次のコネクターで構成されます。
MirrorSourceConnector-
ソースコネクターは、トピックをソースクラスターからターゲットクラスターにレプリケーションします。また、ACL をレプリケーションし、
MirrorCheckpointConnectorを実行する必要があります。 MirrorCheckpointConnector- チェックポイントコネクターは定期的にオフセットを追跡します。有効にすると、ソースクラスターとターゲットクラスター間のコンシューマーグループオフセットも同期されます。
MirrorHeartbeatConnector- ハートビートコネクターは、ソースクラスターとターゲットクラスター間の接続を定期的にチェックします。
以下の表は、コネクタープロパティーと、これらを使用するために設定するコネクターを説明しています。
| プロパティー | sourceConnector | checkpointConnector | heartbeatConnector |
|---|---|---|---|
| ✓ | ✓ | ✓ |
| ✓ | ✓ | ✓ |
| ✓ | ✓ | ✓ |
| ✓ | ✓ | |
| ✓ | ✓ | |
| ✓ | ✓ | |
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ | ||
| ✓ |
9.2.1. コンシューマーグループオフセットの場所の変更トピック
MirrorMaker 2 は、内部トピックを使用してコンシューマーグループのオフセットを追跡します。
offset-syncsトピック-
offset-syncsトピックは、レプリケーションされたトピックパーティションのソースおよびターゲットオフセットをレコードメタデータからマッピングします。 checkpointsトピック-
checkpointsトピックは、各コンシューマーグループでレプリケーションされたトピックパーティションのソースおよびターゲットクラスターで、最後にコミットされたオフセットをマッピングします。
これらは MirrorMaker 2 によって内部的に使用されるため、これらのトピックと直接対話することはありません。
MirrorCheckpointConnector は、オフセット追跡用の チェックポイント を発行します。checkpoints トピックのオフセットは、設定によって事前に決定された間隔で追跡されます。両方のトピックは、フェイルオーバー時に正しいオフセットの位置からレプリケーションの完全復元を可能にします。
offset-syncs トピックの場所は、デフォルトで source クラスターです。offset-syncs.topic.location コネクター設定を使用して、これを target クラスターに変更することができます。トピックが含まれるクラスターへの読み取り/書き込みアクセスが必要です。ターゲットクラスターを offset-syncs トピックの場所として使用すると、ソースクラスターへの読み取りアクセス権しかない場合でも、MirrorMaker 2 を使用できるようになります。
9.2.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 クラスターへ切り替えられると、一部のメッセージが重複する可能性があります。
Java で作成されたアプリケーションがある場合は、RemoteClusterUtils.java ユーティリティーを使用して、アプリケーションを通じてオフセットを同期できます。ユーティリティーは、checkpoints トピックからコンシューマーグループのリモートオフセットを取得します。
9.2.3. ハートビートコネクターを使用するタイミングの決定
ハートビートコネクターはハートビートを出力して、ソース Kafka クラスターとターゲット Kafka クラスター間の接続を確認します。内部 heartbeat トピックはソースクラスターからレプリケートされます。つまり、ハートビートコネクターがソースクラスターに接続されている必要があります。heartbeat トピックはターゲットクラスターに配置されているため、次のことが可能になります。
- データのミラーリング元のすべてのソースクラスターを特定します。
- ミラーリングプロセスの稼働状況と遅延を確認する
これは、プロセスが何らかの理由でスタックしたり停止したりしていないことを確認するのに役立ちます。ハートビートコネクターは、Kafka クラスター間のミラーリングプロセスを監視するための貴重なツールですが、必ずしも使用する必要があるわけではありません。たとえば、デプロイメントのネットワーク遅延が低い場合、またはトピックの数が少ない場合は、ログメッセージやその他の監視ツールを使用してミラーリングプロセスを監視することが推奨されます。ハートビートコネクターを使用しない場合は、MirrorMaker 2 設定からハートビートコネクターを省略してください。
9.2.4. MirrorMaker 2 コネクターの設定の調整
MirrorMaker 2 コネクターが正しく動作することを確認するには、コネクター全体で特定の設定を調整してください。具体的には、次のプロパティーが該当するすべてのコネクターで同じ値であることを確認してください。
-
replication.policy.class -
replication.policy.separator -
offset-syncs.topic.location -
topic.filter.class
たとえば、replication.policy.class の値は、ソース、チェックポイント、およびハートビートコネクターで同じである必要があります。設定が一致していないか欠落していると、データレプリケーションやオフセット同期で問題が発生するため、関連するすべてのコネクターを同じ設定にしておくことが重要です。
