8.2.2. 分散 Kafka Connect でのコネクターの設定

HTTP REST インターフェイス

分散 Kafka Connect のコネクターは、HTTP REST インターフェイスを使用して設定されます。REST インターフェイスはデフォルトで 8083 番ポートをリッスンします。以下のエンドポイントをサポートします。

GET /connectors
既存のコネクターのリストを返します。
POST /connectors
コネクターを作成します。リクエストボディーは、コネクター設定が含まれる JSON オブジェクトである必要があります。
GET /connectors/<name>
特定のコネクターの情報を取得します。
GET /connectors/<name>/config
特定のコネクターの設定を取得します。
PUT /connectors/<name>/config
特定のコネクターの設定を更新します。
GET /connectors/<name>/status
特定のコネクターのステータスを取得します。
PUT /connectors/<name>/pause
コネクターとそのすべてのタスクを一時停止します。コネクターはメッセージの処理を停止します。
PUT /connectors/<name>/resume
一時停止されたコネクターを再開します。
POST /connectors/<name>/restart
コネクターに障害が発生した場合に、コネクターを再起動します。
DELETE /connectors/<name>
コネクターを削除します。
GET /connector-plugins
サポートされるすべてのコネクタープラグインのリストを取得します。

コネクター設定

ほとんどの設定オプションはコネクター固有で、コネクターのドキュメントに含まれています。以下のフィールドは、すべてのコネクターで共通しています。

name
コネクターの名前。特定の Kafka Connect インスタンス内で一意である必要があります。
connector.class
コネクタープラグインのクラス。たとえば、org.apache.kafka.connect.file.FileStreamSinkConnector です。
tasks.max
このコネクターによって使用されるタスクの最大数。タスクは、コネクターが作業を並列処理するために使用します。コネクターは、指定された数よりも少ないタスクを作成する場合があります。
key.converter
メッセージキーを Kafka 形式との間で変換するために使用されるクラス。これにより、Kafka Connect 設定によって設定されたデフォルト値がオーバーライドされます。たとえば、org.apache.kafka.connect.json.JsonConverter です。
value.converter
メッセージペイロードを Kafka 形式との間で変換するために使用されるクラス。これにより、Kafka Connect 設定によって設定されたデフォルト値がオーバーライドされます。たとえば、org.apache.kafka.connect.json.JsonConverter です。

さらに、シンクコネクターには、以下のオプションの 1 つを設定する必要があります。

topics
入力として使用されるトピックのコンマ区切りリスト。
topics.regex
入力として使用されるトピックの Java 正規表現。

その他のオプションについては、特定のコネクターのドキュメントを参照してください。

AMQ Streams には、コネクター設定ファイルのサンプルが含まれています。AMQ Streams インストールディレクトリーの config/connect-file-sink.properties および config/connect-file-source.properties にあります。