EventBridge:RabbitMQ コネクタを使用した RabbitMQ データの同期

前提条件

• EventBridge を有効化し、必要な権限を付与していること。

• ApsaraMQ for RabbitMQ リソースを作成していること。

• OSS サービスを有効化し、バケットを作成していること。 詳細については、「コンソールでのバケットの作成」をご参照ください。

• SAE サービスを有効化していること。 詳細については、「準備」をご参照ください。

• VPC と vSwitch を作成していること。

• ApsaraMQ for Kafka インスタンスを購入し、デプロイしていること。 詳細については、「インスタンスの購入とデプロイ」をご参照ください。

ステップ 1：ApsaraMQ for RabbitMQリソースの作成

1. ApsaraMQ for RabbitMQ コンソールにログインし、インスタンスを作成します。 詳細については、「インスタンスの作成」をご参照ください。

2. インスタンスをクリックして詳細ページを開き、次のリソースを作成します。

   1. ナビゲーションウィンドウで [静的ユーザー名/パスワード] をクリックし、[ユーザー名/パスワードの作成] をクリックします。 詳細については、「ステップ2：リソースの作成」をご参照ください。

      ユーザーを作成した後、ユーザー名とパスワードを保存します。 [インスタンス ID]、[アクセスキー]、[ユーザー名]、[パスワード]、[作成時刻] などの情報を含む新しい認証情報がリストに表示されます。 [操作] 列の [削除] をクリックして、認証情報を削除できます。

   2. ナビゲーションウィンドウで [Vhost リスト] をクリックし、[Vhost の作成] をクリックします。 詳細については、「ステップ2：リソースの作成」をご参照ください。

   3. ナビゲーションウィンドウで [キューリスト] をクリックします。 [現在の Vhost] の横にある [変更] ドロップダウンリストから作成した Vhost を選択し、[キューの作成] をクリックします。 詳細については、「ステップ2：リソースの作成」をご参照ください。

ステップ 2：コネクタの作成

1. RabbitMQ コネクタファイルをダウンロードし、作成した OSS バケットにアップロードします。 詳細については、「ファイルのアップロード」をご参照ください。

2. ApsaraMQ for Kafka コンソールにログインし、概要 ページの リソースの分布 セクションでリージョンを選択します。

3. 左側のナビゲーションペインで、Connector 統合 > タスクリストを選択します。

4. タスクリスト ページで、タスクリストの作成 をクリックします。

5. タスクの作成 パネルで、Task Name と以下のパラメーターを設定します。

   • タスクの作成

     1. Source (ソース) 設定ウィザードで、データプロバイダー で Apache Kafka Connect を選択し、次へ をクリックします。

     2. コネクタの設定 ウィザードで、次のパラメーターを設定し、次へ をクリックします。

        項目	パラメーター	説明
        Kafka Connect プラグイン	[バケット]	OSS バケットを選択します。
        	[ファイル]	アップロードした .zip ファイルを選択します。
        ApsaraMQ for Kafka リソース情報	[Kafka パラメータ設定]	Source Connect を選択します。
        	[Kafka インスタンス]	前提条件で作成したインスタンスを選択します。
        	[Virtual Private Cloud (VPC)]	インスタンスのデプロイに基づいて自動的に入力されます。 この値は変更できません。
        	[vSwitches]	インスタンスのデプロイに基づいて自動的に入力されます。 この値は変更できません。
        	[Security Group]	セキュリティグループを選択します。
        Kafka Connect 設定情報	現在のZIPパッケージ内のプロパティファイルを解析	properties ファイルの作成 を選択します。.ZIP ファイルに含まれている SourceConnector の .properties ファイルを選択します。パスは /etc/source-xxx.properties です。入力ボックスで、次のフィールドの値を更新します。 connector.class：コネクタの Java クラス。 この値は変更しないでください。 tasks.max：タスクの最大数。 rabbitmq.host: RabbitMQ インスタンスの VPC エンドポイントを入力します。エンドポイントは、インスタンスの詳細 ページの エンドポイント情報 セクションで確認できます。 rabbitmq.username：ステップ 1：ApsaraMQ for RabbitMQリソースの作成で作成した ApsaraMQ for RabbitMQ インスタンスの静的ユーザー名を指定します。 rabbitmq.password：ステップ 1：ApsaraMQ for RabbitMQリソースの作成で作成した ApsaraMQ for RabbitMQ インスタンスの静的パスワードを指定します。 rabbitmq.virtual.host：ステップ 1：ApsaraMQ for RabbitMQリソースの作成で作成した Vhost を指定します。 kafka.topic：宛先の Kafka トピック。 データを転送する前に、このトピックを作成する必要があります。 rabbitmq.queue：ステップ 1：ApsaraMQ for RabbitMQリソースの作成で作成したキューを指定します。 次のコードに例を示します。 connector.class=com.ibm.eventstreams.connect.rabbitmqsource.RabbitMQSourceConnector name=rabbitmq-source-connector # ApsaraMQ for RabbitMQ インスタンスの VPC エンドポイント。 rabbitmq.host=xxx # ApsaraMQ for RabbitMQ インスタンスの静的パスワード。 rabbitmq.password=xxx # ApsaraMQ for RabbitMQ インスタンスの静的ユーザー名。 rabbitmq.username=xxx # ApsaraMQ for RabbitMQ インスタンスの Vhost。 rabbitmq.virtual.host=xxx # 宛先 Kafka トピック。 kafka.topic=xxx # ApsaraMQ for RabbitMQ インスタンスのキュー。 rabbitmq.queue=xxx tasks.max=4

     3. インスタンス設定 ウィザードで、次のパラメーターを設定し、次へ をクリックします。

        項目	パラメーター	説明
        ワーカー仕様	[ Worker 仕様]	ワーカー仕様を選択します。
        	[Worker の最小数]	ワーカーの最小数を設定します。
        	[Worker の最大数]	ワーカーの最大数を設定します。 この値はタスクの最大数を超えることはできません。
        	[水平スケーリングしきい値 (%)]	スケーリングは、CPU およびメモリ使用率が設定されたしきい値を上回るか下回るとトリガーされます。Worker の最小数とWorker の最大数の値が等しくない場合にのみ有効です。
        Kafka Connect ワーカー設定	[Worker の依存関係を自動的に作成]	このチェックボックスを選択することを推奨します。 システムは、選択した Kafka インスタンスに Kafka Connect が必要とする内部トピックとコンシューマーグループを自動的に作成し、必要な設定を自動入力します。 これらには以下が含まれます。 オフセットトピック：ソースデータのオフセットを格納します。 トピック名は connect-eb-offset-<task_name> の形式になります。 Config トピック：コネクタとタスクの設定を格納します。 トピック名は connect-eb-config-<task_name> の形式になります。 ステータストピック：コネクタとタスクのステータス情報を格納します。 トピック名は connect-eb-status-<task_name> の形式になります。 Kafka Connect コンシューマーグループ：Kafka Connect ワーカーが内部トピックからメッセージを消費するために使用するコンシューマーグループです。 コンシューマーグループ名は connect-eb-cluster-<task_name> の形式になります。 Kafka ソースコネクタコンシューマーグループ：このパラメーターはシンクコネクタにのみ有効です。 ソーストピックからデータを消費するために使用します。 コンシューマーグループ名は connector-eb-cluster-<task_name>-<connector_name> の形式になります。

     4. 実行設定 セクションで、Log Shipping 方法を Log Service への送信 または Kafka への送信 に設定し、ロールの付与 カードで Connect 依存関係の ロール設定 を設定し、次に 保存 をクリックします。

        重要

        タスクの失敗を防ぐために、AliyunSAEFullAccess 権限を持つロールを選択することを推奨します。

   • タスクプロパティ

     このタスクのリトライポリシーとデッドレターキューを設定します。 詳細については、「リトライとデッドレターキュー」をご参照ください。

   タスクのステータスが実行中になるまで待ちます。これでコネクタは正常に実行されています。

ステップ 3：コネクタのテスト

1. ApsaraMQ for RabbitMQ コンソールにログインし、左側のナビゲーションバーでインスタンスリスト を選択します。

2. インスタンスリスト ページで、トップメニューバーからリージョンを選択し、インスタンスリストで対象のインスタンス名をクリックします。

3. 左側のナビゲーションバーで[キューリスト]をクリックし、対象のキューのActions列にある詳細をクリックします。

4. [キュー詳細] ページで、[宛先情報としてバインド] タブをクリックし、[バインディングの追加] をクリックします。

5. [バインディングの追加] パネルで、[ソースエクスチェンジ] を amq.direct に設定し、OK をクリックします。

6. [Bindings] タブで、amq.direct Exchange の Actions 列にある [メッセージの送信] をクリックし、Kafka のターゲット Topic にメッセージを送信します。 詳細については、「メッセージを送信する」をご参照ください。 ターゲットキューの詳細ページで、[メッセージの送信] をクリックします。 [メッセージの送信] ダイアログボックスで、[routingKey] を 1 に、[messageId] を 123 に、[messageBody] を this is a message に設定します。 [詳細設定] セクションで、値が 12345 の [userId] プロパティを追加し、[OK] をクリックします。

7. ApsaraMQ for Kafka コンソールにログオンし、インスタンス ページで、対象のインスタンスをクリックします。

8. ターゲットインスタンスのページで、ターゲット Topic をクリックし、次に メッセージ検索 をクリックして、挿入されたメッセージデータを表示します。 ターゲット Topic の [メッセージのクエリ] タブで、[クエリ方法] を [時間によるクエリ] に設定し、[パーティション] と [時点] を選択して、[クエリ] をクリックします。 クエリ結果の Value 列に ApsaraMQ for RabbitMQ から送信されたメッセージ内容 (this is a message など) が表示された場合、Connector のテストが成功したことを示します。

All tasks under connector rabbitmq-source-connector failed, please check the error trace of the task.

Kafka connect exited! Please check the error log /opt/kafka/logs/connect.log on sae application to find out the reason why kafka connect exited and update the event streaming with valid arguments to solve it.

Start or update connector xxx failed. Error code=400. Error message=Connector configuration is invalid and contains the following 1 error(s):
Missing required configuration "rabbitmq.host" which has no default value.
You can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`

curl -i -X PUT -H "Accept:application/json" -H "Content-Type:application/json" -d @$CONNECTOR_PROPERTIES_MAPPING http://localhost:8083/connector-plugins/com.ibm.eventstreams.connect.rabbitmqsource.RabbitMQSourceConnector/config/validate

"value":{
    "name":"rabbitmq.host",
    "value":null,
    "recommended_values":[

    ],
    "errors":[
        "Missing required configuration \"rabbitmq.host\" which has no default value."
    ],
    "visible":true
}