1. Home
2. Platform
3. マルチ DC デプロイのアーキテクチャ
4. マルチデーセンターレプリケーション用の Replicator

Docker での Replicator デモ

このデモでは、データセンター間でデータの双方向コピーを行う Confluent Replicator の 2 つのインスタンスを含む、アクティブ/アクティブのマルチデータセンター設計をデプロイします。

この Docker 環境は、本稼働環境ではなくデモのみを目的としたものです。

デモのセットアップ

前提条件:

• Docker
  • Docker バージョン 1.11 またはそれ以降が インストールされ動作している。
  • Docker Compose が インストール済みである 。Docker Compose は、Docker for Mac ではデフォルトでインストールされます。
  • Docker メモリーに最小でも 8 GB が割り当てられている。Docker Desktop for Mac を使用しているとき、Docker メモリーの割り当てはデフォルトで 2 GB です。Docker でデフォルトの割り当てを 8 GB に変更できます。Preferences、Resources、Advanced の順に移動します。
• Git
• インターネットに接続されている
• Confluent Platform で現在サポートされる オペレーティングシステム
• Docker でのネットワークと Kafka

設計

これは、データセンター dc1 と dc2 間でデータの双方向コピーを行う Confluent Replicator の 2 つのインスタンスを含む、アクティブ/アクティブのマルチデータセンター設計です。

Confluent Control Center は、クラスターを管理およびモニタリングするために実行されています。

Confluent Platform コンポーネントのポート

	dc1	dc2
ブローカー	9091	9092
ZooKeeper	2181	2182
Schema Registry	8081（プライマリ）	8082（セカンダリ）
Connect	8381（dc2→dc1 にコピー）	8382（dc1→dc2 にコピー）
Control Center		9021

データ生成とトピック名

データ生成 Docker コンテナー	送信元 DC	送信元トピック	Replicator インスタンス	送信先 DC	送信先トピック
datagen-dc1- topic1	dc1	topic1、_schemas	replicator-dc1 -to-dc2-topic1	dc2	topic1、_schemas
datagen-dc1- topic2	dc1	topic2	replicator-dc1 -to-dc2-topic2	dc2	topic2.r eplica
datagen-dc2- topic1	dc2	topic1	replicator-dc2 -to-dc1-topic1	dc1	topic1

Confluent Replicator （バージョン 5.0.1 以降）は、メッセージヘッダー内の来歴情報を使用して dc1 topic1 と dc2 topic1 の間のデータ循環反復を回避します。

デモの実行

環境

このデモは、以下のソフトウェアで動作が検証されています。

• Docker バージョン 17.06.1-ce
• Docker Compose ファイル形式 2.1 の Docker Compose バージョン 1.14.0

サービスの起動

1. confluentinc/examples GitHub リポジトリのクローンを作成し、6.0.6-post ブランチをチェックアウトします。

   git clone https://github.com/confluentinc/examples
   cd examples
   git checkout 6.0.6-post
   

2. multi-datacenter サンプルディレクトリに移動します。

   cd multi-datacenter/
   

3. 各データセンターのすべてのサービスとトピックのサンプルメッセージを開始します。

   ./start.sh
   

4. docker-compose.yml ファイル内の Kafka ブローカー、Schema Registry、および Connect ワーカーの完全な構成を表示します。

   cat docker-compose.yml
   

マルチデータセンターセットアップの検証

Replicator

このマルチデータセンター環境には、2 つの Apache Kafka® クラスター、dc1 と dc2 があります。これらのクラスターは Confluent Control Center によって管理されています。Confluent Replicator は、dc1 と dc2 間の双方向でデータをコピーしますが、その仕組みをわかりやすく説明するために、以下のセクションでは dc1 から dc2 へのレプリケーションのみについて検討します。

1. デモを起動したら（ 前のセクション を参照）、Chrome ブラウザーで Control Center UI（http://localhost:9021）に移動して、左側のメニューで 2 つの Kafka クラスター、dc1 と dc2 が存在することを確認します。

   

2. このデモでは、2 つの Kafka Connect クラスター、connect-dc1 と connect-dc2 を使用します。Replicator は、通常、送信先 Connect クラスターで実行されるソースコネクターです。そのため、connect-dc1 は dc2 から dc1 にコピーする Replicator を実行し、connect-dc2 は dc1 から dc2 にコピーする Replicator を実行します。バージョン 5.2 以降、Control Center は複数の Kafka Connect クラスターを管理できますが、このデモでは、connect-dc2 で実行され、dc1 から dc2 にコピーする Replicator インスタンスのみを取り上げます。

3. dc1 から dc2 にコピーする Replicator の場合: Connect メニューに移動して、Kafka Connect（connect-dc2 に構成されています）が Replicator の 2 つのインスタンス、replicator-dc1-to-dc2-topic1 と replicator-dc1-to-dc2-topic2 を実行していることを確認します。

   

トピックの検査

1. 各データセンターについて、さまざまなトピックのデータ、来歴情報、タイムスタンプ情報、およびクラスター ID を検査します。

   ./read-topics.sh
   

Replicator のパフォーマンスのモニタリング

Replicator のモニタリングは、以下のために重要です。

1. 複数のデータセンター間のデータ同期状態の測定
2. ネットワークと Confluent Platform のパフォーマンス最適化

Replicator モニタリング

Control Center では、Replicators メニューから Replicator の詳細なモニタリングを確認できます。これを選択すると、現在クラスターで実行されている Replicator インスタンスの概要が表示されます。Replicator ごとに、以下の重要なメトリクスが表示されます。

• Source Cluster - Replicator のレプリケート元になるクラスター
• Source Topics - レプリケートされているトピックの数
• Messages Replicated in/s - 1 秒あたりにレプリケートされたメッセージの数
• Message Lag - 送信元クラスターに存在し、送信先にまだレプリケートされていないメッセージの数
• Latency - メッセージが送信先に存在する時刻と送信元に存在する時刻の差異

ここに表示される Replicator インスタンスを選択すると、特定のインスタンスに関する詳細が表示されます。Status タブでは前述のメトリクスが Connect タスクごとに表示され、Source Topics タブではそれらがトピックごとに表示されます。

さらに Throughput カードを選択すると、詳しいメッセージラグが Connect タスクごとに表示されます。表示されているいずれかのタスクを選択すると、そのタスクがレプリケートしているパーティションのメトリクスが表示されます。

ストリームモニタリング

Control Center のモニタリング機能には、ストリームパフォーマンスのモニタリングが含まれています。これにより、すべてのデータが消費されていること、さらにスループットとレイテンシを検証できます。モニタリングは、コンシューマーグループまたはトピック単位で表示されます。Confluent Replicator は、送信元クラスターからデータを読み取る組み込みのコンシューマーを備えているため、Control Center でそのパフォーマンスをモニタリングできます。Replicator のストリームモニタリングを有効にするには、モニタリングコンシューマーインターセプター でこれを構成します。例 を参照してください。

1. 左側のメニューから dc1 Kafka クラスターを選択し、次に Data Streams を選択します。2 つのコンシューマーグループが存在し、それぞれ dc1 から dc2 へと実行されている Replicator インスタンス、replicator-dc1-to-dc2-topic1 と replicator-dc1-to-dc2-topic2 のものであることを確認します。

   

2. 左側のメニューから dc2 Kafka クラスターを選択し、次に Data Streams を選択します。dc2 から dc1 へと実行されている 1 つのコンシューマーグループ、replicator-dc2-to-dc1-topic1 が存在することを確認します。

   

コンシューマーラグ

Replicator には送信元クラスターからデータを読み取る組み込みのコンシューマーがあり、Connect ワーカーのプロデューサーが送信先クラスターにデータをコミットした後にのみ、そのオフセットをコミットします（コミットの頻度は offset.flush.interval.ms パラメーターで構成します）。送信元クラスターで Replicator の組み込みコンシューマーのコンシューマーラグをモニタリングできます（dc1 から dc2 にデータをコピーする Replicator インスタンスの場合、送信元クラスターは dc1 です）。Replicator のコンシューマーラグをモニタリングする機能は、offset.topic.commit=true が設定されている場合に有効になります（デフォルトは true）。これにより、Replicator は、送信先クラスターにメッセージが書き込まれた後に自身のコンシューマーオフセットを送信元クラスター dc1 にコミットします。

1. dc1 から dc2 にコピーする Replicator の場合 : 左側のメニューから dc1 （送信元クラスター）を選択し、次に Consumers を選択します。2 つのコンシューマーグループが存在し、それぞれ dc1 から dc2 へと実行されている Replicator インスタンス、replicator-dc1-to-dc2-topic1 と replicator-dc1-to-dc2-topic2 のものであることを確認します。Replicator のコンシューマーラグ情報は Control Center および kafka-consumer-groups で確認できますが、JMX を通じて取得することはできません。

   1. replicator-dc1-to-dc2-topic1 をクリックして、topic1 と _schemas の読み取り時の Replicator のコンシューマーラグを表示します（これは docker-compose exec broker-dc1 kafka-consumer-groups --bootstrap-server broker-dc1:29091 --describe --group replicator-dc1-to-dc2-topic1 に相当します）。

      

   2. replicator-dc1-to-dc2-topic2 をクリックして、topic2 の読み取り時の Replicator のコンシューマーラグを表示します（これは docker-compose exec broker-dc1 kafka-consumer-groups --bootstrap-server broker-dc1:29091 --describe --group replicator-dc1-to-dc2-topic2 に相当します）。

      

2. dc1 から dc2 にコピーする Replicator の場合 : 送信先クラスター dc2 で Replicator コンシューマーラグのモニタリングを試行しないでください。Control Center では dc2 内のトピック（つまり topic1、_schemas、topic2.replica）のコンシューマーラグも表示されますが、これは Replicator がそれらから消費していることを意味しません。dc2 でこのコンシューマーラグが表示されるのは、デフォルトで Replicator に offset.timestamps.commit=true が構成されているためです。これにより、Replicator は送信先クラスター dc2 の __consumer_offsets トピックにそのコンシューマーグループの自身のオフセットタイムスタンプをコミットします。ディザスターリカバリが発生した場合、これにより Replicator はセカンダリクラスターに切り替えたときでも中止したポイントから再開できるようになります。

3. コンシューマーラグと、Replicator の組み込みコンシューマーに関連付けられている MBean 属性 records-lag を混同しないでください。この属性は、Replicator の組み込みコンシューマーが元のデータ生成レートを維持できているかどうかを反映しますが、送信先クラスターにメッセージを生成するときのレプリケーションラグは考慮していません。records-lag はリアルタイムであり、通常は値が 0.0 です。

   docker-compose exec connect-dc2 kafka-run-class kafka.tools.JmxTool --object-name "kafka.consumer:type=consumer-fetch-manager-metrics,partition=0,topic=topic1,client-id=replicator-dc1-to-dc2-topic1-0" --attributes "records-lag" --jmx-url service:jmx:rmi:///jndi/rmi://connect-dc2:9892/jmxrmi
   

フェイルオーバー後のアプリケーションの再開

障害が発生したら、Java コンシューマーアプリケーションを別のデータセンターに切り替えます。送信先クラスターでは、アプリケーションは送信元クラスターで中止したポイントからデータの消費を自動的に再開できます。

この機能を使用するには、コンシューマータイムスタンプインターセプター で Java コンシューマーアプリケーションを構成します。このサンプルコードは こちら で公開されています。

1. この Docker 環境の開始後に（ 前のセクション を参照）、コンシューマーを実行して dc1 Kafka クラスターに接続します。コンシューマーグループ ID が自動的に java-consumer-topic1 として構成され、コンシューマータイムスタンプインターセプターとモニタリングインターセプターが使用されます。

   mvn clean package
   mvn exec:java -Dexec.mainClass=io.confluent.examples.clients.ConsumerMultiDatacenterExample -Dexec.args="topic1 localhost:9091 http://localhost:8081 localhost:9092"
   

2. コンシューマー出力で、dc1 と dc2 の両方からデータを読み取っていることを確認します。

   ...
   key = User_1, value = {"userid": "User_1", "dc": "dc1"}
   key = User_9, value = {"userid": "User_9", "dc": "dc2"}
   key = User_6, value = {"userid": "User_6", "dc": "dc2"}
   ...
   

3. コンシューマーが dc1 から消費している場合でも、コンシューマーグループ java-consumer-topic1 に対してコミットされた dc2 コンシューマーオフセットが存在します。以下のコマンドを実行して、dc2 の __consumer_offsets トピックから読み取ります。

   docker-compose exec broker-dc2 kafka-console-consumer --topic __consumer_offsets --bootstrap-server localhost:9092 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" | grep java-consumer
   

4. コミットされたオフセットが存在することを確認します。

   ...
   [java-consumer-topic1,topic1,0]::OffsetAndMetadata(offset=1142, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1547146285084, expireTimestamp=None)
   [java-consumer-topic1,topic1,0]::OffsetAndMetadata(offset=1146, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1547146286082, expireTimestamp=None)
   [java-consumer-topic1,topic1,0]::OffsetAndMetadata(offset=1150, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1547146287084, expireTimestamp=None)
   ...
   

5. Kafka クライアントで Confluent Monitoring Interceptor が構成されている場合、これらはメタデータを _confluent-monitoring というトピックに書き込みます。Kafka クライアントには、Apache Kafka クライアント API を使用して Kafka ブローカーに接続するアプリケーション（カスタムクライアントコードなど）、または組み込みのプロデューサーまたはコンシューマーを備えたサービス（Kafka Connect、ksqlDB、Kafka Streams アプリケーションなど）が含まれています。Control Center はそのトピックを使用して、すべてのメッセージが配信されていることを確認し、スループットとレイテンシのパフォーマンスに関する統計を提供します。その同じトピックから、どのプロデューサーがどのトピックに書き込んでいるか、どのコンシューマーがどのトピックから読み取っているかも確認できます。また、リポジトリにはサンプルの スクリプト が含まれています。

   ./map_topics_clients.py
   

   注釈

   このスクリプトはデモ専用です。本稼働環境には適していません。

6. Java コンシューマーが実行されている安定した状態では、以下のように表示されます。

   Reading topic _confluent-monitoring for 60 seconds...please wait
   
   __consumer_timestamps
     producers
       consumer-1
       producer-10
       producer-11
       producer-6
       producer-8
     consumers
       replicator-dc1-to-dc2-topic1
       replicator-dc1-to-dc2-topic2
       replicator-dc2-to-dc1-topic1
   
   _schemas
     producers
       connect-worker-producer-dc2
     consumers
       replicator-dc1-to-dc2-topic1
   
   topic1
     producers
       connect-worker-producer-dc1
       connect-worker-producer-dc2
       datagen-dc1-topic1
       datagen-dc2-topic1
     consumers
       java-consumer-topic1
       replicator-dc1-to-dc2-topic1
       replicator-dc2-to-dc1-topic1
   
   topic2
     producers
       datagen-dc1-topic2
     consumers
       replicator-dc1-to-dc2-topic2
   
   topic2.replica
     producers
       connect-worker-producer-dc2
   

7. dc1 をシャットダウンします。

   docker-compose stop connect-dc1 schema-registry-dc1 broker-dc1 zookeeper-dc1
   

8. コンシューマーを停止してから再起動して、dc2 Kafka クラスターに接続します。同じコンシューマーグループ ID、java-consumer-topic1 を引き続き使用するため、中止したポイントから再開できます。

   mvn exec:java -Dexec.mainClass=io.confluent.examples.clients.ConsumerMultiDatacenterExample -Dexec.args="topic1 localhost:9092 http://localhost:8082 localhost:9092"
   

9. dc2 からのデータのみが表示されることを確認します。

   ...
   key = User_8, value = {"userid": "User_8", "dc": "dc2"}
   key = User_9, value = {"userid": "User_9", "dc": "dc2"}
   key = User_5, value = {"userid": "User_5", "dc": "dc2"}
   ...
   

デモの停止

デモを終了する場合は、デモを停止して Docker コンテナーとイメージを削除します。

1. 以下のスクリプトを実行してすべてのサービスを停止し、Docker コンテナーとイメージを削除します。

   ./stop.sh
   

2. 以下のコマンドを実行して、実行されているコンテナーが存在しないことを確認します。

   docker container ls
   

次のステップ

• 障害シナリオの発生時の回復性を高めるように複数の Apache Kafka クラスターを設計および構成するための実用的なガイドについては、ディザスターリカバリに関するホワイトペーパー を参照してください。このホワイトペーパーでは、フェイルオーバーとフェイルバックを行い、最終的にリカバリを成功させるための計画について説明しています。
• データレプリケーションに Confluent Platform を使用する方法の概要については、「概要」を参照してください。
• Replicator を構成して独自のマルチクラスターデプロイをセットアップする方法のクイックスタートについては、「チュートリアル: クラスターの境界を越えたデータのレプリケーション」を参照してください。
• Replicator の概要については、「Replicator の概要」を参照してください。
• Confluent Platform を使用してフォロワー、オブザーバー、レプリカ配置を含むストレッチクラスターを作成する方法については、「Multi-Region Clusters」を参照してください。