6.4. Kafka Connect のデプロイ


Kafka Connect は、Apache Kafka と外部システムとの間でデータをストリーミングするツールです。

AMQ Streams では、Kafka Connect は分散 (distributed) モードでデプロイされます。Kafka Connect はスタンドアロンモードでも動作しますが、AMQ Streams ではサポートされません。

Kafka Connect は、コネクター の概念を使用し、スケーラビリティーと信頼性を維持しながら Kafka クラスターで大量のデータを移動するフレームワークを提供します。

Kafka Connect は通常、Kafka を外部データベース、ストレージシステム、およびメッセージングシステムと統合するために使用されます。

Cluster Operator は、KafkaConnect リソースを使用してデプロイされた Kafka Connect クラスターと、KafkaConnector リソースを使用して作成されたコネクターを管理します。

次の手順は、Kafka Connect をデプロイし、ストリーミングデータ用のコネクターを設定する方法を示しています。

注記

コネクター という用語は、Kafka Connect クラスター内で実行されているコネクターインスタンスや、コネクタークラスと同じ意味で使用されます。本ガイドでは、本文の内容で意味が明確である場合に コネクター という用語を使用します。

6.4.1. Kafka Connect の OpenShift クラスターへのデプロイ

この手順では、Cluster Operator を使用して Kafka Connect クラスターを OpenShift クラスターにデプロイする方法を説明します。

Kafka Connect クラスターは、設定可能なノード数 (このノードの別称: ワーカー) を指定して Deployment として実装されます。このノードは、メッセージフローのスケーラビリティーと信頼性が高くなるように、コネクターのワークロードを タスク として分散します。

デプロイメントでは、YAML ファイルの仕様を使って KafkaConnect リソースが作成されます。

AMQ Streams には、設定ファイルのサンプル が用意されています。この手順では、以下のサンプルファイルを使用します。

  • examples/connect/kafka-connect.yaml

手順

  1. Kafka Connect を OpenShift クラスターにデプロイします。examples/connect/kafka-connect.yaml ファイルを使用して Kafka Connect をデプロイします。

    oc apply -f examples/connect/kafka-connect.yaml
  2. デプロイメントのステータスを確認します。

    oc get deployments -n <my_cluster_operator_namespace>

    デプロイメント名と準備状態が表示されている出力

    NAME                        READY  UP-TO-DATE  AVAILABLE
    my-connect-cluster-connect  1/1    1           1

    my-connect-cluster は、Kafka Connect クラスターの名前です。

    READY は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE 出力に 1 が表示されれば、デプロイメントは成功しています。

6.4.2. 複数インスタンスの Kafka Connect 設定

Kafka Connect のインスタンスを複数実行している場合は、以下の config プロパティーのデフォルト設定を変更する必要があります。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect
spec:
  # ...
  config:
    group.id: connect-cluster 
1

    offset.storage.topic: connect-cluster-offsets 
2

    config.storage.topic: connect-cluster-configs 
3

    status.storage.topic: connect-cluster-status  
4

    # ...
# ...
1
Kafka 内の Kafka Connect クラスター ID。
2
コネクターオフセットを保存する Kafka トピック。
3
コネクターおよびタスクステータスの設定を保存する Kafka トピック。
4
コネクターおよびタスクステータスの更新を保存する Kafka トピック。
注記

group.id が同じすべての Kafka Connect インスタンスで、これら 3 つのトピックの値を揃える必要があります。

デフォルト設定を変更しない限り、同じ Kafka クラスターに接続する Kafka Connect インスタンスはそれぞれ同じ値でデプロイされます。事実上、すべてのインスタンスが結合されてクラスターで実行されて同じトピックを使用するようになります。

複数の Kafka Connect クラスターが同じトピックの使用を試みると、Kafka Connect は想定どおりに動作せず、エラーが生成されます。

複数の Kafka Connect インスタンスを実行する場合は、インスタンスごとにこれらのプロパティーの値を変更してください。

6.4.3. コネクタープラグインでの Kafka Connect の拡張

Kafka Connect は、コネクターインスタンスを使用して他のシステムと統合し、データをストリーミングします。コネクターは、次のいずれかのタイプにすることができます。

  • データを Kafka にプッシュするソースコネクター
  • Kafka からデータを抽出するシンクコネクター

このセクションの手順では、次のいずれかを実行してコネクターを追加する方法について説明します。

重要

Kafka Connect REST API または KafkaConnector カスタムリソースを使用 して、直接コネクターの設定を作成します。

独自のコネクターを使用するか、サンプルの FileStreamSourceConnector および FileStreamSinkConnector コネクターを試して、ファイルベースのデータを Kafka クラスターに出し入れすることができます。サンプルファイルコネクターを KafkaConnector リソースとしてデプロイする方法は、「サンプル KafkaConnector リソースのデプロイ」 を参照してください。

注記

Apache Kafka 3.1.0 までは、Kafka Connect の AMQ Streams コンテナーイメージにサンプルファイルコネクターが含まれていました。Apache Kafka 3.1.1 および 3.2.0 以降、これらのコネクターは含まれなくなり、他のコネクターと同様にデプロイする必要があります。

6.4.3.1. AMQ Streams を使用した新しいコンテナーイメージの自動作成

この手順では、AMQ Streams が追加のコネクターで新しいコンテナーイメージを自動的にビルドするように Kafka Connect を設定する方法を説明します。コネクタープラグインは、KafkaConnect カスタムリソースの .spec.build.plugins プロパティーを使用して定義します。AMQ Streams はコネクタープラグインを自動的にダウンロードし、新しいコンテナーイメージに追加します。コンテナーは、.spec.build.output に指定されたコンテナーリポジトリーにプッシュされ、Kafka Connect デプロイメントで自動的に使用されます。

前提条件

イメージをプッシュ、保存、およびプルできる独自のコンテナーレジストリーを提供する必要があります。AMQ Streams は、プライベートコンテナーレジストリーだけでなく、QuayDocker Hub などのパブリックレジストリーもサポートします。

手順

  1. .spec.build.output でコンテナーレジストリーを、.spec.build.plugins で追加のコネクターを指定して、KafkaConnect カスタムリソースを設定します。

    apiVersion: kafka.strimzi.io/v1beta2
    kind: KafkaConnect
    metadata:
      name: my-connect-cluster
    spec: 
    1
    
      #...
      build:
        output: 
    2
    
          type: docker
          image: my-registry.io/my-org/my-connect-cluster:latest
          pushSecret: my-registry-credentials
        plugins: 
    3
    
          - name: debezium-postgres-connector
            artifacts:
              - type: tgz
                url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/1.3.1.Final/debezium-connector-postgres-1.3.1.Final-plugin.tar.gz
                sha512sum: 962a12151bdf9a5a30627eebac739955a4fd95a08d373b86bdcea2b4d0c27dd6e1edd5cb548045e115e33a9e69b1b2a352bee24df035a0447cb820077af00c03
          - name: camel-telegram
            artifacts:
              - type: tgz
                url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.7.0/camel-telegram-kafka-connector-0.7.0-package.tar.gz
                sha512sum: a9b1ac63e3284bea7836d7d24d84208c49cdf5600070e6bd1535de654f6920b74ad950d51733e8020bf4187870699819f54ef5859c7846ee4081507f48873479
      #...
    1
    2
    (必須) 新しいイメージがプッシュされるコンテナーレジストリーの設定。
    3
    (必須) 新しいコンテナーイメージに追加するコネクタープラグインとそれらのアーティファクトの一覧。各プラグインは、1 つ以上の artifact を使用して設定する必要があります。
  2. リソースを作成または更新します。

    $ oc apply -f KAFKA-CONNECT-CONFIG-FILE
  3. 新しいコンテナーイメージがビルドされ、Kafka Connect クラスターがデプロイされるまで待ちます。
  4. Kafka Connect REST API または KafkaConnector カスタムリソースを使用して、追加したコネクタープラグインを使用します。

6.4.3.2. Kafka Connect ベースイメージからの Docker イメージの作成

この手順では、カスタムイメージを作成し、/opt/kafka/plugins ディレクトリーに追加する方法を説明します。

Red Hat Ecosystem Catalog の Kafka コンテナーイメージは、追加のコネクタープラグインで独自のカスタムイメージを作成するためのベースイメージとして使用できます。

AMQ Stream バージョンの Kafka Connect は起動時に、/opt/kafka/plugins ディレクトリーに含まれるサードパーティーのコネクタープラグインをロードします。

手順

  1. registry.redhat.io/amq7/amq-streams-kafka-33-rhel8:2.3.0 をベースイメージとして使用して、新規の Dockerfile を作成します。

    FROM registry.redhat.io/amq7/amq-streams-kafka-33-rhel8:2.3.0
    USER root:root
    COPY ./my-plugins/ /opt/kafka/plugins/
    USER 1001

    プラグインファイルの例

    $ tree ./my-plugins/
    ./my-plugins/
    ├── debezium-connector-mongodb
    │   ├── bson-3.4.2.jar
    │   ├── CHANGELOG.md
    │   ├── CONTRIBUTE.md
    │   ├── COPYRIGHT.txt
    │   ├── debezium-connector-mongodb-0.7.1.jar
    │   ├── debezium-core-0.7.1.jar
    │   ├── LICENSE.txt
    │   ├── mongodb-driver-3.4.2.jar
    │   ├── mongodb-driver-core-3.4.2.jar
    │   └── README.md
    ├── debezium-connector-mysql
    │   ├── CHANGELOG.md
    │   ├── CONTRIBUTE.md
    │   ├── COPYRIGHT.txt
    │   ├── debezium-connector-mysql-0.7.1.jar
    │   ├── debezium-core-0.7.1.jar
    │   ├── LICENSE.txt
    │   ├── mysql-binlog-connector-java-0.13.0.jar
    │   ├── mysql-connector-java-5.1.40.jar
    │   ├── README.md
    │   └── wkb-1.0.2.jar
    └── debezium-connector-postgres
        ├── CHANGELOG.md
        ├── CONTRIBUTE.md
        ├── COPYRIGHT.txt
        ├── debezium-connector-postgres-0.7.1.jar
        ├── debezium-core-0.7.1.jar
        ├── LICENSE.txt
        ├── postgresql-42.0.0.jar
        ├── protobuf-java-2.6.1.jar
        └── README.md

    注記

    この例では、MongoDB、MySQL、および PostgreSQL 用の Debezium コネクターを使用します。Kafka Connect で実行されている Debezium は、他の Kafka Connect タスクと同じように表示されます。

  2. コンテナーイメージをビルドします。
  3. カスタムイメージをコンテナーレジストリーにプッシュします。
  4. 新しいコンテナーイメージを示します。

    以下のいずれかを行います。

    • KafkaConnect カスタムリソースの KafkaConnect.spec.image プロパティーを編集します。

      設定された場合、このプロパティーによって Cluster Operator の STRIMZI_KAFKA_CONNECT_IMAGES 変数がオーバーライドされます。

      apiVersion: kafka.strimzi.io/v1beta2
      kind: KafkaConnect
      metadata:
        name: my-connect-cluster
      spec: 
      1
      
        #...
        image: my-new-container-image 
      2
      
        config: 
      3
      
          #...
      1
      2
      Pod の Docker イメージ。
      3
      Kafka Connect ワーカー (コネクターではない) の設定。

      または

    • install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml ファイルの STRIMZI_KAFKA_CONNECT_IMAGES 変数を編集して新しいコンテナーイメージを参照するようにし、Cluster Operator を再インストールします。

6.4.4. コネクターの作成および管理

コネクタープラグインのコンテナーイメージを作成したら、Kafka Connect クラスターにコネクターインスタンスを作成する必要があります。その後、稼働中のコネクターインスタンスを設定、監視、および管理できます。

コネクターは特定の コネクタークラス のインスタンスで、関連のある外部システムとメッセージについて通信する方法を認識しています。コネクターは多くの外部システムで使用でき、独自のコネクターを作成することもできます。

ソース および シンク タイプのコネクターを作成できます。

ソースコネクター
ソースコネクターは、外部システムからデータを取得し、それをメッセージとして Kafka に提供するランタイムエンティティーです。
シンクコネクター
シンクコネクターは、Kafka トピックからメッセージを取得し、外部システムに提供するランタイムエンティティーです。

6.4.4.1. コネクターの作成および管理用の API

AMQ Streams では、コネクターの作成および管理に 2 つの API が提供されます。

  • KafkaConnector カスタムリソース (別称: KafkaConnectors)
  • Kafka Connect REST API

API を使用すると、以下を行うことができます。

  • コネクターインスタンスのステータスの確認。
  • 稼働中のコネクターの再設定。
  • コネクターインスタンスのコネクタータスク数の増減。
  • コネクターの再起動。
  • 失敗したタスクを含むコネクタータスクの再起動。
  • コネクターインスタンスの一時停止。
  • 一時停止したコネクターインスタンスの再開。
  • コネクターインスタンスの削除。

KafkaConnector カスタムリソース

KafkaConnectors を使用すると、Kafka Connect のコネクターインスタンスを OpenShift ネイティブに作成および管理できるため、cURL などの HTTP クライアントは必要ありません。他の Kafka リソースと同様に、KafkaConnector YAML ファイルで、コネクターの目的の状態を宣言して、このファイルを OpenShift クラスターにデプロイしてコネクターインスタンスを作成します。KafkaConnector リソースは、リンク先の Kafka Connect クラスターと同じ namespace にデプロイする必要があります。

該当する KafkaConnector リソースを更新して稼働中のコネクターインスタンスを管理した後、更新を適用します。該当する KafkaConnector を削除して、コネクターを削除します。

下位バージョンの AMQ Streams との互換性を維持するため、KafkaConnectors はデフォルトで無効になっています。Kafka Connect クラスターの KafkaConnectors を有効にするには、KafkaConnect リソースで strimzi.io/use-connector-resources アノテーションを true に設定します。手順は、Kafka Connect の設定 を参照してください。

KafkaConnectors が有効になると、Cluster Operator によって監視が開始されます。KafkaConnectors に定義された設定と一致するよう、稼働中のコネクターインスタンスの設定を更新します。

AMQ Streams には KafkaConnector 設定ファイルの例が含まれます。これをもとに、Kafka Connector のデプロイを使用して FileStreamSourceConnectorFileStreamSinkConnector を作成して管理できます

注記

KafkaConnector リソースにアノテーションを付けて、コネクターの再起動 または コネクタータスクの再起動 ができます。

Kafka Connect API

Kafka Connect REST API でサポートされる操作は、Apache Kafka Connect API のドキュメント で説明されています。

Kafka Connect API から KafkaConnectors の使用への切り替え

Kafka Connect API から KafkaConnectors の使用に切り替えると、コネクターを管理できます。スイッチの作成は、次の作業を以下の順序で行います。

  1. 設定で KafkaConnector リソースをデプロイし、コネクターインスタンスを作成します。
  2. strimzi.io/use-connector-resources アノテーションを true に設定して、Kafka Connect 設定で KafkaConnectors を有効にします。
警告

リソースを作成する前に KafkaConnectors を有効にすると、すべてのコネクターが削除されます。

KafkaConnectors から Kafka Connect API の使用に切り替えるには、まず Kafka Connect 設定から KafkaConnectors を有効にするアノテーションを削除します。それ以外の場合、Kafka Connect REST API を使用して直接行われた手動による変更は、 Cluster Operator によって元に戻されます。

6.4.4.2. サンプル KafkaConnector リソースのデプロイ

KafkaConnector リソースは、Cluster Operator によるコネクターの管理に OpenShift ネイティブのアプローチを提供します。AMQ Streams には、設定ファイルのサンプル が用意されています。この手順では、examples/connect/source-connector.yaml ファイルを使用して、次のコネクターインスタンスを KafkaConnector リソースとして作成します。

  • Kafka ライセンスファイル (ソース) から各行を読み取り、データをメッセージとして単一の Kafka トピックに書き込む FileStreamSourceConnector インスタンス。
  • Kafka トピックからメッセージを読み取り、メッセージを一時ファイル (シンク) に書き込む FileStreamSinkConnector インスタンス。

または、examples/connect/kafka-connect-build.yaml ファイルを使用して、ファイルコネクターを使用して新しい Kafka Connect イメージを構築することもできます。

Apache Kafka 3.1.0 までは、サンプルファイルコネクタープラグインが Apache Kafka に含まれていました。Apache Kafka の 3.1.1 および 3.2.0 リリースから、例を他のコネクターと同様にプラグインパスに追加する必要があります。詳細は、コネクタープラグインを使用した Kafka Connect の拡張 を参照してください。

注記

「コネクタープラグインでの Kafka Connect の拡張」 で説明されているように、実稼働環境では、必要な Kafka Connect コネクターを使用してコンテナーイメージを準備します。

FileStreamSourceConnector および FileStreamSinkConnector が例として提供されています。ここで記載されているようなコンテナーで上記のコネクターを実行することは、実稼働のユースケースとして適切ではありません。

前提条件

手順

  1. examples/connect/source-connector.yaml ファイルを編集します。

    apiVersion: kafka.strimzi.io/v1beta2
    kind: KafkaConnector
    metadata:
      name: my-source-connector 
    1
    
      labels:
        strimzi.io/cluster: my-connect-cluster 
    2
    
    spec:
      class: org.apache.kafka.connect.file.FileStreamSourceConnector 
    3
    
      tasksMax: 2 
    4
    
      config: 
    5
    
        file: "/opt/kafka/LICENSE" 
    6
    
        topic: my-topic 
    7
    
        # ...
    1
    コネクターの名前として使用される KafkaConnector リソースの名前。OpenShift リソースで有効な名前を使用します。
    2
    コネクターインスタンスを作成する Kafka Connect クラスターの名前。コネクターは、リンク先の Kafka Connect クラスターと同じ namespace にデプロイする必要があります。
    3
    コネクタークラスのフルネームまたはエイリアス。これは、Kafka Connect クラスターによって使用されているイメージに存在するはずです。
    4
    コネクターが作成できる Kafka Connect Tasks の最大数。
    5
    キーと値のペアとしての コネクター設定
    6
    このサンプルソースコネクター設定では、/opt/kafka/LICENSE ファイルからデータが読み取られます。
    7
    ソースデータのパブリッシュ先となる Kafka トピック。
  2. OpenShift クラスターでソース KafkaConnector を作成します。

    oc apply -f examples/connect/source-connector.yaml
  3. examples/connect/sink-connector.yaml ファイルを作成します。

    touch examples/connect/sink-connector.yaml
  4. 以下の YAML を sink-connector.yaml ファイルに貼り付けます。

    apiVersion: kafka.strimzi.io/v1beta2
    kind: KafkaConnector
    metadata:
      name: my-sink-connector
      labels:
        strimzi.io/cluster: my-connect
    spec:
      class: org.apache.kafka.connect.file.FileStreamSinkConnector 
    1
    
      tasksMax: 2
      config: 
    2
    
        file: "/tmp/my-file" 
    3
    
        topics: my-topic 
    4
    1
    コネクタークラスのフルネームまたはエイリアス。これは、Kafka Connect クラスターによって使用されているイメージに存在するはずです。
    2
    キーと値のペアとしての コネクター設定
    3
    ソースデータのパブリッシュ先となる一時ファイル。
    4
    ソースデータの読み取り元となる Kafka トピック。
  5. OpenShift クラスターにシンク KafkaConnector を作成します。

    oc apply -f examples/connect/sink-connector.yaml
  6. コネクターリソースが作成されたことを確認します。

    oc get kctr --selector strimzi.io/cluster=MY-CONNECT-CLUSTER -o name
    
    my-source-connector
    my-sink-connector

    MY-CONNECT-CLUSTER を Kafka Connect クラスターに置き換えます。

  7. コンテナーで、kafka -console-consumer.sh を実行して、ソースコネクターによってトピックに書き込まれたメッセージを読み取ります。

    oc exec MY-CLUSTER-kafka-0 -i -t -- bin/kafka-console-consumer.sh --bootstrap-server MY-CLUSTER-kafka-bootstrap.NAMESPACE.svc:9092 --topic my-topic --from-beginning
ソースおよびシンクコネクターの設定オプション

コネクター設定は、KafkaConnector リソースの spec.config プロパティーで定義されます。

FileStreamSourceConnector クラスおよび FileStreamSinkConnector クラスは、Kafka Connect REST API と同じ設定オプションをサポートします。他のコネクターは異なる設定オプションをサポートします。

Expand
表6.1 FileStreamSource コネクタークラスの設定オプション
名前タイプデフォルト値説明

file

String

Null

メッセージを書き込むソースファイル。指定のない場合は、標準入力が使用されます。

topic

List

Null

データのパブリッシュ先となる Kafka トピック。

Expand
表6.2 FileStreamSinkConnector クラスの設定オプション
名前タイプデフォルト値説明

file

String

Null

メッセージを書き込む宛先ファイル。指定のない場合は標準出力が使用されます。

topics

List

Null

データの読み取り元となる 1 つ以上の Kafka トピック。

topics.regex

String

Null

データの読み取り元となる 1 つ以上の Kafka トピックと一致する正規表現。

6.4.4.3. Kafka コネクターの再起動の実行

この手順では、OpenShift アノテーションを使用して Kafka コネクターの再起動を手動でトリガーする方法を説明します。

前提条件

  • Cluster Operator が稼働中である。

手順

  1. 再起動する Kafka コネクターを制御する KafkaConnector カスタムリソースの名前を見つけます。

    oc get KafkaConnector
  2. コネクターを再起動するには、OpenShift で KafkaConnector リソースにアノテーションを付けます。たとえば、oc annotate を使用すると以下のようになります。

    oc annotate KafkaConnector KAFKACONNECTOR-NAME strimzi.io/restart=true
  3. 次の調整が発生するまで待ちます (デフォルトでは 2 分ごとです)。

    アノテーションが調整プロセスで検出されれば、Kafka コネクターは再起動されます。Kafka Connect が再起動リクエストを受け入れると、アノテーションは KafkaConnector カスタムリソースから削除されます。

6.4.4.4. Kafka コネクタータスクの再起動の実行

この手順では、OpenShift アノテーションを使用して Kafka コネクタータスクの再起動を手動でトリガーする方法を説明します。

前提条件

  • Cluster Operator が稼働中である。

手順

  1. 再起動する Kafka コネクタータスクを制御する KafkaConnector カスタムリソースの名前を見つけます。

    oc get KafkaConnector
  2. KafkaConnector カスタムリソースから再起動するタスクの ID を検索します。タスク ID は 0 から始まる負の値ではない整数です。

    oc describe KafkaConnector KAFKACONNECTOR-NAME
  3. コネクタータスクを再起動するには、OpenShift で KafkaConnector リソースにアノテーションを付けます。たとえば、oc annotate を使用してタスク 0 を再起動します。

    oc annotate KafkaConnector KAFKACONNECTOR-NAME strimzi.io/restart-task=0
  4. 次の調整が発生するまで待ちます (デフォルトでは 2 分ごとです)。

    アノテーションが調整プロセスで検出されれば、Kafka コネクタータスクは再起動されます。Kafka Connect が再起動リクエストを受け入れると、アノテーションは KafkaConnector カスタムリソースから削除されます。

6.4.4.5. Kafka Connect API の公開

KafkaConnector リソースを使用してコネクターを管理する代わりに、Kafka Connect REST API を使用します。Kafka Connect REST API は、<connect_cluster_name>-connect-api:8083 で実行しているサービスとして利用できます。ここで、<connect_cluster_name> は、お使いの Kafka Connect クラスターの名前になります。サービスは、Kafka Connect インスタンスの作成時に作成されます。

注記

strimzi.io/use-connector-resources アノテーションは KafkaConnectors を有効にします。アノテーションを KafkaConnect リソース設定に適用した場合、そのアノテーションを削除して Kafka Connect API を使用する必要があります。それ以外の場合、Kafka Connect REST API を使用して直接行われた手動による変更は、 Cluster Operator によって元に戻されます。

コネクター設定を JSON オブジェクトとして追加できます。

コネクター設定を追加するための curl 要求の例

curl -X POST \
  http://my-connect-cluster-connect-api:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{ "name": "my-source-connector",
    "config":
    {
      "connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",
      "file": "/opt/kafka/LICENSE",
      "topic":"my-topic",
      "tasksMax": "4",
      "type": "source"
    }
}'

API には OpenShift クラスター内でのみアクセスできます。OpenShift クラスター外部で実行しているアプリケーションに Kafka Connect API がアクセスできるようにする場合は、以下の機能のいずれかを使用して Kafka Connect API を手動で公開できます。

  • LoadBalancer または NodePort タイプのサービス
  • Ingress リソース
  • OpenShift ルート
注記

接続は安全ではないため、外部からのアクセスはよく考えてから許可してください。

サービスを作成する場合には、<connect_cluster_name>-connect-api サービスの selector からラベルを使用して、サービスがトラフィックをルーティングする Pod を設定します。

サービスのセレクター設定

# ...
selector:
  strimzi.io/cluster: my-connect-cluster 
1

  strimzi.io/kind: KafkaConnect
  strimzi.io/name: my-connect-cluster-connect 
2

#...

1
OpenShift クラスターでの Kafka Connect カスタムリソースの名前。
2
Cluster Operator によって作成された Kafka Connect デプロイメントの名前。

また、外部クライアントからの HTTP 要求を許可する NetworkPolicy を作成する必要もあります。

Kafka Connect API への要求を許可する NetworkPolicy の例

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: my-custom-connect-network-policy
spec:
  ingress:
  - from:
    - podSelector: 
1

        matchLabels:
          app: my-connector-manager
    ports:
    - port: 8083
      protocol: TCP
  podSelector:
    matchLabels:
      strimzi.io/cluster: my-connect-cluster
      strimzi.io/kind: KafkaConnect
      strimzi.io/name: my-connect-cluster-connect
  policyTypes:
  - Ingress

1
API への接続が許可される Pod のラベル。

クラスター外でコネクター設定を追加するには、curl コマンドで API を公開するリソースの URL を使用します。

6.4.4.6. Kafka Connect API へのアクセスの制限

Kafka Connect API へのアクセスを信頼できるユーザーのみに制限して、不正なアクションや潜在的なセキュリティーの問題を防ぐことが重要です。Kafka Connect API は、コネクター設定を変更するための広範な機能を提供するため、セキュリティー対策を講じることがさらに重要になります。管理者により安全であると想定されている機密情報が、Kafka Connect API にアクセスできるユーザーに取得されてしまう可能性があります。

Kafka Connect REST API には、OpenShift クラスターへのアクセスが認証されており、ホスト名/IP アドレス、ポート番号など、エンドポイント URL を知っている場合には、アクセスできます。

たとえば、組織が Kafka Connect クラスターとコネクターを使用して、機密データを顧客データベースから中央データベースにストリーミングするとします。管理者は設定プロバイダープラグインを使用して、顧客データベースと中央データベースへの接続に関連する機密情報 (データベース接続の詳細や認証情報など) を保存します。設定プロバイダーは、この機密情報が許可されていないユーザーに公開されるのを防ぎます。ただし、Kafka Connect API にアクセスできるユーザーは、管理者の同意なしに顧客データベースにアクセスできます。これを行うには、偽のデータベースをセットアップし、それに接続するコネクターを設定します。次に、顧客データベースを参照するようにコネクター設定を変更しますが、データを中央データベースに送信する代わりに、偽のデータベースに送信します。偽のデータベースに接続するようにコネクターを設定すると、設定プロバイダーにセキュアに保存されているにもかかわらず、顧客データベースに接続するためのログインの詳細と認証情報が傍受されます。

KafkaConnector カスタムリソースを使用している場合、デフォルトでは、OpenShift RBAC ルールにより、OpenShift クラスター管理者のみがコネクターに変更を加えることが許可されます。AMQ Streams リソースを管理するクラスター管理者以外のユーザーを指定 することもできます。Kafka Connect 設定で KafkaConnector リソースを有効にすると、Kafka Connect REST API を使用して直接行われた変更は Cluster Operator によって元に戻されます。KafkaConnector リソースを使用していない場合、デフォルトの RBAC ルールは Kafka Connect API へのアクセスを制限しません。OpenShift RBAC を使用して Kafka Connect REST API への直接アクセスを制限する場合は、KafkaConnector リソースを有効にして使用する必要があります。

セキュリティーを強化するために、Kafka Connect API の次のプロパティーを設定することを推奨します。

connector.client.config.override.policy

connector.client.config.override.policy プロパティーを None (デフォルト) に設定して、コネクター設定が Kafka Connect 設定とそれが使用するコンシューマーおよびプロデューサーをオーバーライドしないようにします。

コネクターオーバーライドポリシーを指定する設定例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  # ...
  config:
    connector.client.config.override.policy: None
# ...

Red Hat logoGithubredditYoutubeTwitter

詳細情報

試用、購入および販売

コミュニティー

会社概要

Red Hat は、企業がコアとなるデータセンターからネットワークエッジに至るまで、各種プラットフォームや環境全体で作業を簡素化できるように、強化されたソリューションを提供しています。

多様性を受け入れるオープンソースの強化

Red Hat では、コード、ドキュメント、Web プロパティーにおける配慮に欠ける用語の置き換えに取り組んでいます。このような変更は、段階的に実施される予定です。詳細情報: Red Hat ブログ.

Red Hat ドキュメントについて

Legal Notice

Theme

© 2026 Red Hat
トップに戻る