10.4. MirrorMaker 2.0 を使用した Kafka クラスター間でのデータの同期

手順

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

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

   clusters=cluster-1,cluster-2 1
   
   cluster-1.bootstrap.servers=<my-cluster>-kafka-bootstrap-<my-project>:443 2
   cluster-1.security.protocol=SSL 3
   cluster-1.ssl.truststore.password=<my-truststore-password>
   cluster-1.ssl.truststore.location=<path-to-truststore>/truststore.cluster-1.jks
   cluster-1.ssl.keystore.password=<my-keystore-password>
   cluster-1.ssl.keystore.location=<path-to-keystore>/user.cluster-1.p12
   
   cluster-2.bootstrap.servers=<my-cluster>-kafka-bootstrap-<my-project>:443 4
   cluster-2.security.protocol=SSL 5
   cluster-2.ssl.truststore.password=<my-truststore-password>
   cluster-2.ssl.truststore.location=<path-to-truststore>/truststore.cluster-2.jks
   cluster-2.ssl.keystore.password=<my-keystore-password>
   cluster-2.ssl.keystore.location=<path-to-keystore>/user.cluster-2.p12
   
   cluster-1->cluster-2.enabled=true 6
   cluster-1->cluster-2.topics=.* 7
   cluster-2->cluster-1.enabled=true 8
   cluster-2->cluster-1B->C.topics=.* 9
   
   replication.policy.separator=- 10
   sync.topic.acls.enabled=false 11
   refresh.topics.interval.seconds=60 12
   refresh.groups.interval.seconds=60 13

   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-1 クラスターから cluster-2 クラスターに複製します。

   8

   cluster-2 クラスターから cluster-1 クラスターへのレプリケーションフローが有効になりました。

   9

   すべてのトピックを cluster-2 クラスターから cluster-1 クラスターに複製します。

   10

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

   11

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

   12

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

   13

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

2. （オプション）必要に応じて、リモートトピックの名前の自動変更を上書きするポリシーを追加します。その名前の前にソースクラスターの名前を追加するのではなく、トピックが元の名前を保持します。

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

   replication.policy.class=io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy

3. ターゲットクラスターで 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

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

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

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

5. ターゲットクラスターごとに、トピックが複製されていることを確認します。

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