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

MirrorMaker 2.0 を使用して、設定を介して Kafka クラスター間のデータを同期します。

以前のバージョンの MirrorMaker は、レガシーモードで MirrorMaker 2.0 を実行 することにより、引き続きサポートされます。

設定では以下を指定する必要があります。

  • 各 Kafka クラスター
  • TLS 認証を含む各クラスターの接続情報

  • レプリケーションのフローおよび方向

    • クラスター対クラスター
    • トピック対トピック
  • レプリケーションルール
  • コミットされたオフセット追跡間隔

この手順では、プロパティーファイルで設定を作成し、MirrorMaker スクリプトファイルを使用して接続を設定する際にプロパティーを渡し、MirrorMaker 2.0 を実装する方法を説明します。

注記

MirrorMaker 2.0 は Kafka Connect を使用して接続を確立し、クラスター間でデータを転送します。Kafka は、データの複製に MirrorMaker シンクおよびソースコネクターを提供します。MirrorMaker スクリプトの代わりにコネクターを使用する場合は、Kafka Connect クラスターでコネクターを設定する必要があります。詳細は、Apache Kafka のドキュメント を参照してください。

作業を開始する前に

設定プロパティーファイルの例は ./config/connect-mirror-maker.properties にあります。

前提条件

  • 複製している各 Kafka クラスターノードのホストに AMQ Streams がインストールされている必要がある。

手順

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

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

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

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

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

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

    /bin/kafka-topics.sh --zookeeper  <ZooKeeperAddress> --list