ApsaraMQ for Kafka: コネクタ管理 - ApsaraMQ for Kafka (Kafka) - Alibaba Cloud - ApsaraMQ for Kafka

Message Queue for Apache Kafka コンソールで、コネクタの構成やログを表示したり、コネクタを一時停止または有効化したり、構成を変更したり、コネクタをテストしたり、消費進捗を確認したり、コネクタを削除したりできます。

前提条件

以下のいずれかのコネクタを作成済みである必要があります。

説明

構成を変更できるのは、Function Compute シンクコネクタおよび MySQL ソースコネクタのみです。

注意事項

構成を変更できるのは、Function Compute シンクコネクタおよび MySQL ソースコネクタのみです。
コネクタは、実行中 (Running) 状態の場合にのみ一時停止できます。また、一時停止 (Suspended) 状態の場合にのみ有効化できます。

コネクタ構成の表示

コネクタの基本情報、ソース情報、送信先情報を表示します。

ApsaraMQ for Kafka コンソールにログインします。
概要ページのリソースの分布セクションで、リージョンを選択します。
左側のナビゲーションウィンドウで、Connector タスクリスト をクリックします。
Connector タスクリスト ページで、インスタンスの選択 ドロップダウンリストから対象のインスタンスを選択し、リストから目的のコネクタを見つけ、操作列の詳細をクリックします。

コネクタ詳細ページでは、基本情報、ソース情報、およびターゲット情報を確認できます。

コネクタログの表示

コネクタの問題をトラブルシューティングする必要がある場合は、Message Queue for Apache Kafka コンソールでそのログを表示できます。

ApsaraMQ for Kafka コンソールにログインします。
概要ページのリソースの分布セクションで、リージョンを選択します。
左側のナビゲーションウィンドウで、Connector タスクリスト をクリックします。
[Connector タスクリスト] ページで、[インスタンスの選択] ドロップダウンリストからコネクタが属するインスタンスを選択します。リストで対象のコネクタを見つけます。[操作] 列で、詳細 > ログの表示 を選択します。

または、目的のコネクタ名をクリックするか、操作列の詳細をクリックします。コネクタ詳細ページで、右上隅のログの表示 をクリックします。

説明
操作列におけるログの表示 の位置は、タスクのステータスによって異なります。実際の UI 表示が優先されます。

コネクタログページでは、コネクタのログを確認できます。

コネクタ構成の変更

Function Compute シンクコネクタまたは MySQL ソースコネクタを作成後、Message Queue for Apache Kafka コンソールでその構成を変更できます。

ApsaraMQ for Kafka コンソールにログインします。
概要ページのリソースの分布セクションで、リージョンを選択します。
左側のナビゲーションウィンドウで、Connector タスクリスト をクリックします。
［Connector タスクリスト］ページで、［インスタンスの選択］ドロップダウンリストからコネクタが属するインスタンスを選択します。リスト内で対象のコネクタを見つけます。［操作］列で、詳細 > 設定の変更を選択します。

または、目的のコネクタ名をクリックするか、操作列の詳細をクリックします。コネクタ詳細ページで、右上隅の設定の変更 をクリックします。

設定の変更 パネルで、必要に応じて構成パラメーターを変更し、OK をクリックします。

Function Compute シンクコネクタの構成変更

パラメーター	説明
コンシューマースレッドの同時発生数	ソーストピックの同時コンシューマースレッド数です。デフォルト値は 6 です。有効値： 1 2 3 6 12
失敗の処理	トピックパーティションへのメッセージ配信が失敗した場合の動作を決定します。有効値：サブスクリプションの継続：エラーが発生したパーティションからのサブスクリプションを継続し、エラーをログに記録します。サブスクリプションの停止：エラーが発生したパーティションからのサブスクリプションを停止し、エラーをログに記録します。説明ログの表示方法の詳細については、「コネクタの管理」をご参照ください。エラーコードに基づく解決策の詳細については、「エラーコード一覧」をご参照ください。
送信モード	メッセージ送信モードです。有効値：同期：送信者と受信者は同時に実行され、通信の準備ができている必要があります。非同期：メッセージ送信後、送信者はサーバーからの応答を待たずに他のタスクを続行できます。
ロットサイズを送信	バッチで送信するメッセージ数です。デフォルト値は 20 です。有効値の範囲は [1, 5000] です。
再試行回数	メッセージ送信が失敗した後のリトライ回数です。デフォルト値は 2 です。有効値の範囲は 1 ～ 3 です。ただし、一部のエラーではメッセージ送信失敗時のリトライがサポートされていません。エラーコードの詳細については、「エラーコード一覧」をご参照ください。エラーコードごとのリトライサポート状況は次のとおりです。 4xx：429 エラーコードを除き、リトライはサポートされません。 5xx：リトライがサポートされます。説明コネクタは、メッセージを Function Compute に送信するためにInvokeFunction API を呼び出します。

MySQL ソースコネクタの構成変更

パラメーター

説明

データベーステーブル

ApsaraDB RDS for MySQL インスタンス内で同期する 1 つ以上のテーブル名です。複数のテーブル名を指定する場合は、カンマ (,) で区切ります。

データベーステーブルと送信先トピックの間には 1 対 1 のマッピングが存在します。

説明

既存のタスクに対してテーブルを追加または変更する場合、あらかじめMessage Queue for Apache Kafka コンソールで対応する Topic を作成しておく必要があります。Topic 名はTopic prefix_table name形式である必要があります。これを行わないと、実行時エラーが発生します。詳細については、「手順 1：Topic の作成」をご参照ください。

変更が完了したら、Connector タスクリスト ページで目的のコネクタを見つけ、操作列の詳細をクリックします。Connector の詳細 ページで、更新されたコネクタ構成を確認できます。

コネクタの一時停止または有効化

Message Queue for Apache Kafka コンソールで、実行中のコネクタを一時的に停止（一時停止）したり、一時停止中のコネクタを再開（有効化）したりできます。

ApsaraMQ for Kafka コンソールにログインします。
概要ページのリソースの分布セクションで、リージョンを選択します。
左側のナビゲーションウィンドウで、Connector タスクリスト をクリックします。
Connector タスクリスト ページで、インスタンスの選択 ドロップダウンリストから対象のインスタンスを選択し、リストから目的のコネクタを見つけ、必要に応じて一時停止または有効化します。
- コネクタを一時停止するには、操作列で詳細 > 一時停止 を選択します。ヒント ダイアログボックスで一時停止を確認し、OK をクリックします。
- コネクタを有効化する場合：操作列で有効にする をクリックします。ヒント ダイアログボックスで起動を確認し、OK をクリックします。
または、目的のコネクタ名をクリックするか、操作列の詳細をクリックします。コネクタ詳細ページで、右上隅の一時停止 または有効にする をクリックします。
Connector タスクリスト ページで、一時停止または有効化したコネクタを見つけます。ステータスが一時停止 または実行中 に変化します。

説明
MySQL ソースコネクタタスクが長時間一時停止されている場合、再起動に失敗することがあります。その場合は、該当タスクを見つけ、操作列のタスクの設定 をクリックします。DataWorks ページで、目的の同期タスクの行を見つけ、開始をクリックします。開始ダイアログボックスでReset Offset を選択し、Start Time Point を設定してから、OK をクリックします。

コネクタのテスト

Message Queue for Apache Kafka コンソールからテストメッセージを送信して、コネクタを検証できます。

ApsaraMQ for Kafka コンソールにログインします。
概要ページのリソースの分布セクションで、リージョンを選択します。
左側のナビゲーションウィンドウで、Connector タスクリスト をクリックします。
Connector タスクリスト ページで、インスタンスの選択 ドロップダウンリストから対象のインスタンスを選択し、リストから目的のコネクタを見つけ、操作列のテスト をクリックします。

または、目的のコネクタ名をクリックするか、操作列の詳細をクリックします。コネクタ詳細ページで、右上隅のテスト をクリックします。

説明
操作列におけるテスト の位置は、タスクのステータスによって異なる場合があります。
メッセージの送信 パネルで、テストメッセージを送信します。
1. メッセージキー テキストボックスに、demo などのメッセージキーを入力します。
2. メッセージの内容 テキストボックスに、{"key": "test"} などのテストメッセージ内容を入力します。
3. 指定されたパーティションに送信 オプションを使用して、メッセージを特定のパーティションに送信するかどうかを制御します。
  - はいをクリックし、パーティション ID テキストボックスに 0 などのパーティション ID を入力します。パーティション ID の確認方法については、「パーティションステータスの表示」をご参照ください。
  - パーティションを指定しない場合は、いいえ をクリックします。

消費進捗の表示

コネクタの実行開始後、ソーストピックに対するそのコンシューマーグループの消費状況を確認できます。コンシューマーグループ名は connect-{task_name} です。

ApsaraMQ for Kafka コンソールにログインします。
概要ページのリソースの分布セクションで、リージョンを選択します。
左側のナビゲーションウィンドウで、Connector タスクリスト をクリックします。
Connector タスクリスト ページで、インスタンスの選択 ドロップダウンリストから対象のインスタンスを選択し、リストから目的のコネクタを見つけ、操作列の消費の進捗状況 をクリックします。

または、目的のコネクタ名をクリックするか、操作列の詳細をクリックします。コネクタ詳細ページで、右上隅の消費の進捗状況 をクリックします。

コンシューマーグループ消費詳細パネルでは、コネクタタスクがソーストピックで使用しているコンシューマーグループの消費詳細を確認できます。

コネクタの削除

各Message Queue for Apache Kafka インスタンスには、コネクタ数の上限があります。不要になったコネクタは、Message Queue for Apache Kafka コンソールで削除できます。

重要

MySQL ソースコネクタタスクが実行中 (Running) 状態の場合、Message Queue for Apache Kafka コンソールで直接削除することはできません。DataWorks コンソールにログインしてコネクタタスクを停止および非公開にした後、チケットを送信してMessage Queue for Apache Kafka コネクタタスクのメタデータを削除する必要があります。その他のコネクタ（Function Compute シンクコネクタ、MaxCompute シンクコネクタ、OSS シンクコネクタ、Elasticsearch シンクコネクタなど）は、Message Queue for Apache Kafka コンソールで直接削除できます。
コネクタを削除すると、自動作成または手動作成に関係なく、そのコネクタに依存する 5 つのトピックと 2 つのコンシューマーグループもシステムにより同時に削除されます。

ApsaraMQ for Kafka コンソールにログインします。
概要ページのリソースの分布セクションで、リージョンを選択します。
左側のナビゲーションウィンドウで、Connector タスクリスト をクリックします。
［Connector タスクリスト］ページで、［インスタンスの選択］ドロップダウンリストからコネクタが属するインスタンスを選択します。リスト内で対象のコネクタを見つけます。［操作］列で、詳細 > 削除を選択します。

または、目的のコネクタ名をクリックするか、操作列の詳細をクリックします。コネクタ詳細ページで、右上隅の削除をクリックします。
ヒント ダイアログボックスで、OK をクリックしてコネクタを削除します。