MirrorMaker 2.0 を使用して Kafka トピックを複製する

このチュートリアルでは、MirrorMaker 2.0 ソースコネクタを使用して、Managed Service for Apache Kafka クラスタ間でデータを複製する方法について説明します。

クラスタは同じプロジェクトに配置することも、異なるプロジェクトに配置することもできます。また、同じリージョンに配置することも、異なるリージョンに配置することもできます。このチュートリアルでは、リージョン間で同じプロジェクト内のレプリケーションを設定します。ただし、他の組み合わせの手順は同じです。

アーキテクチャ

このシナリオでは、次の 3 つのクラスタがあります。

  • データが複製されている Kafka クラスタ。このクラスタはデータのソースであるため、ソースクラスタと呼ばれます。

  • 複製されたデータが書き込まれる Kafka クラスタ。このクラスタは、ターゲット クラスタと呼ばれます。

  • MirrorMaker 2.0 ソース コネクタを作成して管理できる Connect クラスタ。

Connect クラスタがターゲット クラスタにアタッチされます。書き込みオペレーションのレイテンシを最小限に抑えるには、Connect クラスタをターゲット クラスタと同じリージョンに配置します。

次の図に、これらのコンポーネントを示します。

Kafka データ レプリケーション シナリオのアーキテクチャ

データ レプリケーションは、MirrorMaker 2.0 ソースコネクタによって実行されます。このシナリオでは、他の 2 つの MirrorMaker 2.0 コネクタは省略可能です。

  • MirrorMaker 2.0 チェックポイント コネクタ: ターゲット クラスタのコンシューマーがソース クラスタと同じポイントから処理を再開できるようにすることで、シームレスなフェイルオーバーを実現します。

  • MirrorMaker 2.0 ハートビート コネクタ: ソース Kafka クラスタで定期的なハートビート メッセージを生成します。これにより、データ複製の状態とステータスをモニタリングできます。

このチュートリアルでは、これらのオプションのコネクタは使用しません。詳細については、MirrorMaker 2.0 を使用するタイミングをご覧ください。

始める前に

コンソール

  1. Google Cloud アカウントにログインします。 Google Cloudを初めて使用する場合は、 アカウントを作成して、実際のシナリオで Google プロダクトのパフォーマンスを評価してください。新規のお客様には、ワークロードの実行、テスト、デプロイができる無料クレジット $300 分も差し上げます。

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  3. Verify that billing is enabled for your Google Cloud project.

  4. Enable the Managed Kafka API, if it is not already enabled.

    Roles required to enable APIs

    To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

    Enable the API

  5. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  6. Verify that billing is enabled for your Google Cloud project.

  7. Enable the Managed Kafka API, if it is not already enabled.

    Roles required to enable APIs

    To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

    Enable the API

    8.

  8. プロジェクトに次のロールがあることを確認します。 マネージド Kafka クラスタ編集者、マネージド Kafka Connect クラスタ編集者、マネージド Kafka コネクタ編集者

    ロールを確認する

    1. Google Cloud コンソールで、[IAM] ページに移動します。

      IAM に移動
    2. プロジェクトを選択します。

    3. [プリンシパル] 列で、自分または自分が所属するグループの行をすべて確認します。所属するグループについては、管理者にお問い合わせください。

    4. 自分のメールアドレスを含む行の [ロール] 列で、ロールのリストに必要なロールが含まれているかどうか確認します。

    ロールを付与する

    1. Google Cloud コンソールで、[IAM] ページに移動します。

      IAM に移動
    2. プロジェクトを選択します。
    3. [ アクセスを許可] をクリックします。

    4. [新しいプリンシパル] フィールドに、ユーザー ID を入力します。 これは通常、Google アカウントのメールアドレスです。

    5. [ロールを選択] をクリックし、ロールを検索します。
    6. 追加のロールを付与するには、 [別のロールを追加] をクリックして各ロールを追加します。
    7. [保存] をクリックします。

gcloud

  1. Google Cloud アカウントにログインします。 Google Cloudを初めて使用する場合は、 アカウントを作成して、実際のシナリオで Google プロダクトのパフォーマンスを評価してください。新規のお客様には、ワークロードの実行、テスト、デプロイができる無料クレジット $300 分も差し上げます。

  2. Google Cloud CLI をインストールします。

  3. 外部 ID プロバイダ（IdP）を使用している場合は、まず フェデレーション ID を使用して gcloud CLI にログインする必要があります。

  4. gcloud CLI を初期化するには、次のコマンドを実行します。

    gcloud init

  5. Google Cloud プロジェクトを作成または選択します。

    プロジェクトの選択または作成に必要なロール

    • プロジェクトを選択する: プロジェクトの選択に特定の IAM ロールは必要ありません。ロールが付与されているプロジェクトであれば、どのプロジェクトでも選択できます。
    • プロジェクトを作成する: プロジェクトを作成するには、resourcemanager.projects.create 権限を含むプロジェクト作成者ロール（roles/resourcemanager.projectCreator）が必要です。詳しくは、ロールを付与する方法をご覧ください。

    • Google Cloud プロジェクトを作成します。

      gcloud projects create PROJECT_ID

      PROJECT_ID は、作成する Google Cloud プロジェクトの名前に置き換えます。

    • 作成した Google Cloud プロジェクトを選択します。

      gcloud config set project PROJECT_ID

      PROJECT_ID は、 Google Cloud プロジェクトの名前に置き換えます。

  6. Google Cloud プロジェクトで課金が有効になっていることを確認します。

  7. Managed Kafka API がまだ有効になっていない場合は、有効にします。

    API を有効にするために必要なロール

    API を有効にするには、serviceusage.services.enable 権限が必要です。プロジェクトを作成した場合は、オーナーロール（roles/owner）を通じてこの権限がすでに付与されている可能性があります。それ以外の場合は、Service Usage 管理者ロール（roles/serviceusage.serviceUsageAdmin）を通じてこの権限を取得できます。ロールを付与する方法を確認する。

    gcloud services enable managedkafka.googleapis.com

  8. Google Cloud CLI をインストールします。

  9. 外部 ID プロバイダ（IdP）を使用している場合は、まず フェデレーション ID を使用して gcloud CLI にログインする必要があります。

  10. gcloud CLI を初期化するには、次のコマンドを実行します。

    gcloud init

  11. Google Cloud プロジェクトを作成または選択します。

    プロジェクトの選択または作成に必要なロール

    • プロジェクトを選択する: プロジェクトの選択に特定の IAM ロールは必要ありません。ロールが付与されているプロジェクトであれば、どのプロジェクトでも選択できます。
    • プロジェクトを作成する: プロジェクトを作成するには、resourcemanager.projects.create 権限を含むプロジェクト作成者ロール（roles/resourcemanager.projectCreator）が必要です。詳しくは、ロールを付与する方法をご覧ください。

    • Google Cloud プロジェクトを作成します。

      gcloud projects create PROJECT_ID

      PROJECT_ID は、作成する Google Cloud プロジェクトの名前に置き換えます。

    • 作成した Google Cloud プロジェクトを選択します。

      gcloud config set project PROJECT_ID

      PROJECT_ID は、 Google Cloud プロジェクトの名前に置き換えます。

  12. Google Cloud プロジェクトで課金が有効になっていることを確認します。

  13. Managed Kafka API がまだ有効になっていない場合は、有効にします。

    API を有効にするために必要なロール

    API を有効にするには、serviceusage.services.enable 権限が必要です。プロジェクトを作成した場合は、オーナーロール（roles/owner）を通じてこの権限がすでに付与されている可能性があります。それ以外の場合は、Service Usage 管理者ロール（roles/serviceusage.serviceUsageAdmin）を通じてこの権限を取得できます。ロールを付与する方法を確認する。

    gcloud services enable managedkafka.googleapis.com
    14.

  14. ユーザー アカウントにロールを付与します。次の IAM ロールごとに次のコマンドを 1 回実行します。 roles/managedkafka.clusterEditor, roles/managedkafka.connectClusterEditor, roles/managedkafka.connectorEditor 

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

    次のように置き換えます。

    • PROJECT_ID: プロジェクト ID。
    • USER_IDENTIFIER: ユーザー アカウントの識別子。例: myemail@example.com
    • ROLE: ユーザー アカウントに付与する IAM ロール。

ソース Kafka クラスタを作成する

このステップでは、Managed Service for Apache Kafka クラスタを作成します。このクラスタは、レプリケートするデータを含むソースクラスタです。

コンソール

  1. [Managed Service for Apache Kafka] > [クラスタ] ページに移動します。

    [クラスタ] に移動

  2. [ 作成] をクリックします。

  3. [クラスタ名] フィールドに、クラスタの名前を入力します。

  4. [リージョン] リストで、クラスタのロケーションを選択します。

  5. [ネットワーク構成] で、クラスタがアクセスできるサブネットを構成します。

    1. [プロジェクト] で、該当するプロジェクトを選択します。
    2. [ネットワーク] で、プロジェクト内の VPC ネットワークを選択します。
    3. [サブネット] でサブネットを選択します。
    4. [完了] をクリックします。

  6. [作成] をクリックします。

クラスタの作成中は、クラスタの状態が Creating になります。クラスタの作成が完了すると、状態は Active になります。

gcloud

移行元の Kafka クラスタを作成するには、managed-kafka clusters create コマンドを実行します。

gcloud managed-kafka clusters create SOURCE_KAFKA_CLUSTER \
--location=SOURCE_REGION \
--cpu=3 \
--memory=3GiB \
--subnets=projects/PROJECT_ID/regions/SOURCE_REGION/subnetworks/SOURCE_SUBNET \
--async

次のように置き換えます。

  • SOURCE_KAFKA_CLUSTER: Kafka クラスタの名前

  • SOURCE_REGION: クラスタのロケーション

    サポートされているロケーションについては、Managed Service for Apache Kafka のロケーションをご覧ください。

  • PROJECT_ID: プロジェクト ID

  • SOURCE_SUBNET: クラスタをデプロイするサブネット（例: default）

このコマンドは非同期で実行され、オペレーション ID を返します。

Check operation [projects/PROJECT_ID/locations/SOURCE_SUBNET/operations/OPERATION_ID] for status.

作成オペレーションの進行状況を追跡するには、gcloud managed-kafka operations describe コマンドを使用します。

gcloud managed-kafka operations describe OPERATION_ID \
  --location=SOURCE_REGION

詳細については、クラスタ作成オペレーションをモニタリングするをご覧ください。

Kafka トピックを作成する

ソース クラスタの準備ができたら、次のようにトピックを作成します。

コンソール

  1. [Managed Service for Apache Kafka] > [クラスタ] ページに移動します。

    [クラスタ] に移動

  2. クラスタの名前をクリックします。

  3. クラスタの詳細ページで、 [トピックを作成] をクリックします。

  4. [トピック名] ボックスに、トピックの名前を入力します。

  5. [作成] をクリックします。

gcloud

Kafka トピックを作成するには、managed-kafka topics create コマンドを実行します。

gcloud managed-kafka topics create TOPIC_NAME \
--cluster=SOURCE_KAFKA_CLUSTER \
--location=SOURCE_REGION \
--partitions=10 \
--replication-factor=3

次のように置き換えます。

  • TOPIC_NAME: 作成する Kafka トピックの名前
  • SOURCE_KAFKA_CLUSTER: ソースクラスタの名前
  • SOURCE_REGION: ソースクラスタを作成したリージョン

ターゲット Kafka クラスタを作成する

このステップでは、2 つ目の Managed Service for Apache Kafka クラスタを作成します。このクラスタは、MirrorMaker 2.0 がデータを複製するターゲット クラスタです。

コンソール

  1. [Managed Service for Apache Kafka] > [クラスタ] ページに移動します。

    [クラスタ] に移動

  2. [ 作成] をクリックします。

  3. [クラスタ名] フィールドに、クラスタの名前を入力します。

  4. [リージョン] リストで、クラスタのロケーションを選択します。ソースクラスタが配置されているリージョンとは異なるリージョンを選択します。

  5. [ネットワーク構成] で、ターゲット クラスタにアクセスできるサブネットを構成します。サブネットは、移行元クラスタのサブネットと同じ VPC ネットワークに存在する必要があります。

    1. [プロジェクト] で、該当するプロジェクトを選択します。
    2. [ネットワーク] で、移行元クラスタのサブネットと同じ VPC ネットワークを選択します。
    3. [サブネット] でサブネットを選択します。
    4. [完了] をクリックします。

  6. [作成] をクリックします。

クラスタの作成中は、クラスタの状態が Creating になります。クラスタの作成が完了すると、状態は Active になります。

gcloud

ターゲット Kafka クラスタを作成するには、managed-kafka clusters create コマンドを実行します。

gcloud managed-kafka clusters create TARGET_KAFKA_CLUSTER \
--location=TARGET_REGION \
--cpu=3 \
--memory=3GiB \
--subnets=projects/PROJECT_ID/regions/TARGET_REGION/subnetworks/TARGET_SUBNET \
--async

次のように置き換えます。

  • TARGET_KAFKA_CLUSTER: Kafka クラスタの名前

  • TARGET_REGION: クラスタのロケーション。ソースクラスタが配置されているリージョンとは異なるリージョンを選択します。

    サポートされているロケーションについては、Managed Service for Apache Kafka のロケーションをご覧ください。

  • PROJECT_ID: プロジェクト ID

  • TARGET_SUBNET: クラスタをデプロイするサブネット（例: default）

    ソース クラスタと同じ VPC 内のサブネットを選択します。

Connect クラスタを作成する

この手順では、Connect クラスタを作成します。通常、Connect クラスタの作成には 20 ～ 30 分かかります。

この手順を開始する前に、前の手順のターゲット Kafka クラスタが完全に作成されていることを確認してください。

コンソール

  1. [Managed Service for Apache Kafka] > [Connect クラスタ] ページに移動します。

    [Connect クラスタ] に移動

  2. [ 作成] をクリックします。

  3. [Connect クラスタ名] に文字列を入力します。my-connect-cluster

  4. [プライマリ Kafka クラスタ] で、前の手順で作成したターゲット Kafka クラスタを選択します。（ソース Kafka クラスタは選択しないでください）。

  5. [ロケーション]、[ネットワーク構成]、[ワーカー サブネット] では、デフォルト値を使用するか、特定の要件に合わせてカスタマイズできます。

  6. Connect クラスタでソースクラスタの DNS ドメインを解決できるようにするには、次の操作を行います。

    1. [解決可能な DNS ドメイン] を開きます。

    2. [DNS ドメインを追加] をクリックします。

    3. [Kafka クラスタ] リストで、移行元の Kafka クラスタを選択します。

  7. [作成] をクリックします。

クラスタの作成中は、クラスタの状態が Creating になります。クラスタの作成が完了すると、状態は Active になります。

gcloud

Connect クラスタを作成するには、gcloud managed-kafka connect-clusters create コマンドを実行します。

gcloud managed-kafka connect-clusters create CONNECT_CLUSTER \
  --location=TARGET_REGION \
  --cpu=12 \
  --memory=12GiB \
  --primary-subnet=projects/PROJECT_ID/regions/TARGET_REGION/subnetworks/TARGET_SUBNET \
  --kafka-cluster=TARGET_KAFKA_CLUSTER \
  --dns-name=SOURCE_KAFKA_CLUSTER.SOURCE_REGION.managedkafka.PROJECT_ID.cloud.goog. \
  --async

次のように置き換えます。

  • CONNECT_CLUSTER: Connect クラスタの名前
  • TARGET_REGION: ターゲット Kafka クラスタを作成したリージョン
  • PROJECT_ID: プロジェクト ID
  • TARGET_SUBNET: ターゲット Kafka クラスタを作成したサブネット
  • TARGET_KAFKA_CLUSTER: ターゲット Kafka クラスタの名前
  • SOURCE_KAFKA_CLUSTER: ソース Kafka クラスタの名前
  • SOURCE_REGION: ソース Kafka クラスタを作成したリージョン

このコマンドは非同期で実行され、オペレーション ID を返します。

Check operation [projects/PROJECT_ID/locations/TARGET_REGION/operations/OPERATION_ID] for status.

作成オペレーションの進行状況を追跡するには、gcloud managed-kafka operations describe コマンドを使用します。

gcloud managed-kafka operations describe OPERATION_ID \
  --location=TARGET_REGION

詳細については、クラスタ作成オペレーションをモニタリングするをご覧ください。

MirrorMaker 2.0 Source コネクタを作成する

このステップでは、MirrorMaker 2.0 ソース コネクタを作成します。このコネクタは、ソース Kafka クラスタからターゲット Kafka クラスタにメッセージを複製します。

コンソール

  1. [Managed Service for Apache Kafka] > [Connect クラスタ] ページに移動します。

    [Connect クラスタ] に移動

  2. Connect クラスタの名前をクリックします。

  3. [コネクタを作成] をクリックします。

  4. [コネクタ名] に文字列を入力します。例: mm2-connector

  5. [コネクタ プラグイン] リストで、[MirrorMaker 2.0 Source] を選択します。

  6. [ソース クラスタ] で、[Managed Service for Apache Kafka クラスタ] を選択します。

  7. [Kafka クラスタ] リストで、ソースクラスタを選択します。

  8. [カンマ区切りのトピック名またはトピックの正規表現] ボックスに、複製する Kafka トピックの名前を入力します。

  9. [作成] をクリックします。

gcloud

  1. gcloud managed-kafka clusters describe コマンドを実行して、移行元 Kafka クラスタのブートストラップ アドレスを取得します。

    gcloud managed-kafka clusters describe SOURCE_KAFKA_CLUSTER \
  --location=SOURCE_REGION \
  --project=PROJECT_ID \
  --format="value(bootstrapAddress)"

    次のように置き換えます。

    • SOURCE_KAFKA_CLUSTER: ソース Kafka クラスタの名前
    • SOURCE_REGION: ソースクラスタのリージョン
    • PROJECT_ID: プロジェクト ID

  2. 前のコマンドを再度実行して、ターゲット Kafka クラスタのブートストラップ アドレスを取得します。

    gcloud managed-kafka clusters describe TARGET_KAFKA_CLUSTER \
  --location=TARGET_REGION \
  --project=PROJECT_ID \
  --format="value(bootstrapAddress)"

    次のように置き換えます。

    • TARGET_KAFKA_CLUSTER: ターゲット Kafka クラスタの名前
    • TARGET_REGION: ターゲット クラスタのリージョン

  3. MirrorMaker 2.0 ソースコネクタを作成するには、gcloud managed-kafka connectors create コマンドを実行します。

    gcloud managed-kafka connectors create CONNECTOR_NAME \
  --location=TARGET_REGION \
  --connect-cluster=CONNECT_CLUSTER \
  --configs=connector.class=org.apache.kafka.connect.mirror.MirrorSourceConnector,\
source.cluster.alias=source,\
source.cluster.bootstrap.servers=SOURCE_CLUSTER_BOOTSTRAP_ADDRESS,\
target.cluster.alias=target,\
target.cluster.bootstrap.servers=TARGET_CLUSTER_BOOTSTRAP_ADDRESS,\
tasks.max=3,\
topics=TOPIC_NAME

    次のように置き換えます。

    • CONNECTOR_NAME: コネクタの名前（mm2-connector など）
    • CONNECT_CLUSTER: Connect クラスタの名前
    • SOURCE_CLUSTER_BOOTSTRAP_ADDRESS: 手順 1 で取得した移行元クラスタのブートストラップ アドレス。
    • TARGET_CLUSTER_BOOTSTRAP_ADDRESS: 手順 2 で取得したターゲット クラスタのブートストラップ アドレス。
    • TOPIC_NAME: レプリケートするトピックの名前。このパラメータでは、トピック名のカンマ区切りのリストまたは正規表現を指定することもできます。

MirrorMaker 2.0 ソースコネクタは、ターゲット クラスタに "source.TOPIC_NAME" という名前の新しいトピックを作成します。ここで、TOPIC_NAME はソース クラスタのトピックの名前です。

結果を表示

メッセージが複製されていることを確認するには、Kafka コマンドライン ツールを使用します。Kafka CLI の設定については、ドキュメントの CLI を使用してメッセージを生成して使用するのクライアント マシンを設定するをご覧ください。

たとえば、ソースクラスタにメッセージを送信するには、コマンドラインで次のコマンドを入力します。

export BOOTSTRAP=$(gcloud managed-kafka clusters describe SOURCE_KAFKA_CLUSTER \
 --location=SOURCE_REGION \
 --project=PROJECT_ID \
 --format="value(bootstrapAddress)"
)

for msg in {1..10}; do
  echo "message $msg"
done | kafka-console-producer.sh --topic TOPIC_NAME \
       --bootstrap-server $BOOTSTRAP --producer.config client.properties

ターゲット クラスタから重複したメッセージを読み取るには、コマンドラインで次のコマンドを入力します。

export BOOTSTRAP=$(gcloud managed-kafka clusters describe TARGET_KAFKA_CLUSTER \
 --location=TARGET_REGION \
 --project=PROJECT_ID \
 --format="value(bootstrapAddress)"
)

kafka-console-consumer.sh --topic source.TOPIC_NAME --from-beginning \
 --bootstrap-server $BOOTSTRAP --consumer.config client.properties

出力は次のようになります。

message 1
message 2
message 3
message 4
[...]

クリーンアップ

このチュートリアルで使用したリソースについて、Google Cloud アカウントに課金されないようにするには、リソースを含むプロジェクトを削除するか、プロジェクトを維持して個々のリソースを削除します。

コンソール

  1. Connect クラスタを削除します。

    1. [Managed Service for Apache Kafka] > [Connect クラスタ] ページに移動します。

      [Connect クラスタ] に移動

    2. Connect クラスタを選択し、[削除] をクリックします。

  2. 移行元の Kafka クラスタを削除します。

    1. [Managed Service for Apache Kafka] > [クラスタ] ページに移動します。

      [クラスタ] に移動

    2. Kafka クラスタを選択して、[削除] をクリックします。

  3. 前の手順を繰り返して、ターゲット Kafka クラスタを削除します。

gcloud

  1. Connect クラスタを削除するには、gcloud managed-kafka connect-clusters delete コマンドを使用します。

    gcloud managed-kafka connect-clusters delete CONNECT_CLUSTER \
  --location=TARGET_REGION --async

  2. ソース Kafka クラスタを削除するには、gcloud managed-kafka clusters delete コマンドを使用します。

    gcloud managed-kafka clusters delete SOURCE_KAFKA_CLUSTER \
  --location=SOURCE_REGION --async

  3. 前の手順を繰り返して、ターゲット Kafka クラスタを削除します。

    gcloud managed-kafka clusters delete TARGET_KAFKA_CLUSTER \
  --location=TARGET_REGION --async

次のステップ