9.6. MirrorMaker 2 を専用モードで実行する

手順

1. テキストエディターでサンプルプロパティーファイルを開くか、新しいプロパティーファイルを作成し、ファイルを編集して接続情報と各 Kafka クラスターのレプリケーションフローを追加します。

   以下の例は、cluster-1 および cluster-2 の 2 つのクラスターを双方向に接続する設定を示しています。クラスター名は、clusters プロパティーで設定できます。

   MirrorMaker 2 の設定例

   clusters=cluster-1,cluster-2 1
   
   cluster-1.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_one>:443 2
   cluster-1.security.protocol=SSL 3
   cluster-1.ssl.truststore.password=<truststore_name>
   cluster-1.ssl.truststore.location=<path_to_truststore>/truststore.cluster-1.jks_
   cluster-1.ssl.keystore.password=<keystore_name>
   cluster-1.ssl.keystore.location=<path_to_keystore>/user.cluster-1.p12
   
   cluster-2.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_two>:443 4
   cluster-2.security.protocol=SSL 5
   cluster-2.ssl.truststore.password=<truststore_name>
   cluster-2.ssl.truststore.location=<path_to_truststore>/truststore.cluster-2.jks_
   cluster-2.ssl.keystore.password=<keystore_name>
   cluster-2.ssl.keystore.location=<path_to_keystore>/user.cluster-2.p12
   
   cluster-1->cluster-2.enabled=true 6
   cluster-2->cluster-1.enabled=true 7
   cluster-1->cluster-2.topics=.* 8
   cluster-2->cluster-1.topics=topic-1, topic-2 9
   cluster-1->cluster-2.groups=.* 10
   cluster-2->cluster-1.groups=group-1, group-2 11
   
   replication.policy.separator=- 12
   sync.topic.acls.enabled=false 13
   refresh.topics.interval.seconds=60 14
   refresh.groups.interval.seconds=60 15

   1

   各 Kafka クラスターは、そのエイリアスで識別されます。

   2

   ブートストラップアドレス およびポート 443 を使用した、cluster-1 の接続情報。両方のクラスターはポート 443 を使用し、OpenShift Routes を使用して Kafka に接続します。

   3

   ssl. プロパティーは、cluster-1 の TLS 設定を定義します。

   4

   cluster-2 の接続情報です。

   5

   ssl. プロパティーは、cluster-2 の TLS 設定を定義します。

   6

   cluster-1 から cluster-2 へのレプリケーションフローが有効になっています。

   7

   cluster-2 から cluster-1. へのレプリケーションフローが有効になっています。

   8

   cluster-1 から cluster-2 へのすべてのトピックのレプリケーションです。ソースコネクターは指定のトピックをレプリケーションします。チェックポイントコネクターは、指定されたトピックのオフセットを追跡します。

   9

   cluster-2 から cluster-1 への特定のトピックのレプリケーション。

   10

   cluster-1 から cluster-2 へのすべてのコンシューマーグループのレプリケーション。チェックポイントコネクターは、指定されたコンシューマーグループをレプリケーションします。

   11

   cluster-2 から cluster-1 への特定のコンシューマーグループのレプリケーション。

   12

   リモートトピック名の変更に使用する区切り文字を定義します。

   13

   有効にすると、同期されたトピックに ACL が適用されます。デフォルトは false です。

   14

   新しいトピックの同期をチェックする間隔です。

   15

   新しいコンシューマーグループの同期をチェックする間隔です。

2. オプション: 必要に応じて、リモートトピックの名前の自動変更をオーバーライドするポリシーを追加します。その名前の前にソースクラスターの名前を追加する代わりに、トピックが元の名前を保持します。

   このオプションの設定は、active/passive バックアップおよびデータ移行に使用されます。

   replication.policy.class=org.apache.kafka.connect.mirror.IdentityReplicationPolicy

3. オプション: コンシューマーグループのオフセットを同期する場合は、設定を追加して同期を有効にし、管理します。

   refresh.groups.interval.seconds=60
   sync.group.offsets.enabled=true 1
   sync.group.offsets.interval.seconds=60 2
   emit.checkpoints.interval.seconds=60 3

   1

   コンシューマーグループのオフセットを同期する任意設定。これは、active/passive 設定でのリカバリーに便利です。同期はデフォルトでは有効になっていません。

   2

   コンシューマーグループオフセットの同期が有効な場合は、同期の頻度を調整できます。

   3

   オフセット追跡のチェック頻度を調整します。オフセット同期の頻度を変更する場合は、これらのチェックの頻度も調整することを推奨します。

4. ターゲットクラスターで ZooKeeper および Kafka を起動します。

   su - kafka
   /opt/kafka/bin/zookeeper-server-start.sh -daemon \
   /opt/kafka/config/zookeeper.properties

   /opt/kafka/bin/kafka-server-start.sh -daemon \
   /opt/kafka/config/server.properties

5. プロパティーファイルで定義したクラスター接続設定およびレプリケーションポリシーで MirrorMaker を起動します。

   /opt/kafka/bin/connect-mirror-maker.sh \
   /opt/kafka/config/connect-mirror-maker.properties

   MirrorMaker はクラスター間の接続を設定します。

6. ターゲットクラスターごとに、トピックがレプリケーションされていることを確認します。

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --list