Streams for Apache Kafka on OpenShift の概要

Kafka の基盤のデータストリーム処理機能とコンポーネントアーキテクチャーによって以下が提供されます。

Kafka の機能は、以下に適しています。

Streams for Apache Kafka は、OpenShift で Kafka を実行するためのコンテナーイメージおよび Operator を提供します。Streams for Apache Kafka Operator は、専門的な OpenShift で Kafka を効果的に管理するための専門的な運用知識をもとに特別に構築されています。

Operator は以下のプロセスを単純化します。

Kafka クラスターは、メッセージ配信を行うブローカーで設定されます。

ZooKeeper はクラスター管理に使用されます。Kafka を KRaft (Kafka Raft メタデータ) モードでデプロイすると、ブローカーとコントローラーのロールが Kafka ノード内に統合されるため、クラスターの管理が簡素化され、ZooKeeper が不要になります。Kafka ノードはブローカー、コントローラー、またはその両方のロールを引き継ぎます。ロールは、ノードプールを使用して Streams for Apache Kafka で設定されます。

他の Kafka コンポーネントはそれぞれ Kafka クラスターと対話し、特定のロールを実行します。

Streams for Apache Kafka の使用経験を最大化するには、Kafka がメッセージブローカーとしてどのように動作するかを理解する必要があります。

プロデューサーおよびコンシューマーは、ブローカー経由でメッセージ (パブリッシュしてサブスクライブ) を送受信します。メッセージは、キー (オプション) と、メッセージデータ、ヘッダー、および関連するメタデータが含まれる 値 で設定されます。キーは、メッセージの件名またはメッセージのプロパティーの特定に使用されます。メッセージはバッチで配信されます。バッチやレコードには、レコードのタイムスタンプやオフセットの位置など、クライアントのフィルタリングやルーティングに役立つ情報を提供するヘッダーとメタデータが含まれます。

Kafka Connect は、コネクターインスタンスを使用して他のシステムと統合し、データをストリーミングします。

Kafka Connect は、起動時に既存のコネクターインスタンスをロードし、データストリーミングタスクとコネクター設定をワーカー Pod 全体に分散します。ワーカーは、コネクターインスタンスのタスクを実行します。各ワーカーは個別の Pod として実行され、Kafka Connect クラスターの耐障害性を高めます。ワーカーよりも多くのタスクがある場合は、ワーカーには複数のタスクが割り当てられます。ワーカーに障害が発生した場合、そのタスクは Kafka Connect クラスター内のアクティブなワーカーに自動的に割り当てられます。

ストリーミングデータで使用される主な Kafka Connect コンポーネントは次のとおりです。

Kafka Bridge は、次のような操作を実行するメソッドを使用して、Kafka クラスターへの HTTP リクエストをサポートします。

上記の方法で、JSON レスポンスと HTTP レスポンスコードのエラー処理を行います。メッセージは JSON またはバイナリー形式で送信できます。

Kafka Bridge を使用して、内部および外部の HTTP クライアントアプリケーションの両方を Kafka クラスターに統合できます。

Streams for Apache Kafka は、Cluster Operator を使用してクラスターをデプロイおよび管理します。デフォルトでは、Streams for Apache Kafka をデプロイすると、単一の Cluster Operator レプリカがデプロイされます。リーダーの選択でレプリカを追加し、中断が発生した場合に追加の Cluster Operator がスタンバイ状態になるようにすることができます。

Cluster Operator は、以下の Kafka コンポーネントのクラスターを管理します。

クラスターは、カスタムリソースを使用してデプロイされます。

たとえば、以下のように Kafka クラスターをデプロイします。

Cluster Operator は、Kafka リソースの設定を通じて、以下の Streams for Apache Kafka Operator をデプロイすることもできます。

Topic Operator および User Operator は、デプロイメントの Entity Operator 内で機能します。

Streams for Apache Kafka Drain Cleaner のデプロイメントで Cluster Operator を使用すると、Pod のエビクションに役立ちます。Streams for Apache Kafka Drain Cleaner をデプロイすることで、Cluster Operator を使用して OpenShift ではなく Kafka Pod を移動できます。Streams for Apache Kafka Drain Cleaner は、エビクトされる Pod にローリング更新のアノテーションを付けます。このアノテーションは、Cluster Operator にローリング更新を実行するように通知します。

Topic Operator は、KafkaTopic リソースを通じて Kafka クラスター内のトピックを管理する方法を提供します。

Topic Operator は、Kafka を記述する KafkaTopic リソースを監視し、それらが Kafka クラスター内で適切に設定されていることを確認することによって、Kafka トピックを管理します。

KafkaTopic が作成、削除、または変更されると、Topic Operator は Kafka トピックに対して対応するアクションを実行します。

アプリケーションのデプロイメントの一部として KafkaTopic を宣言すると、Topic Operator が Kafka トピックを管理します。

Topic Operator は次のモードで動作します。

User Operator は、KafkaUser リソースを通じて Kafka クラスター内のユーザーを管理する方法を提供します。

User Operator は、Kafka ユーザーが記述される KafkaUser リソースを監視して Kafka クラスターの Kafka ユーザーを管理し、Kafka ユーザーが Kafka クラスターで適切に設定されるようにします。

KafkaUser が作成、削除、または変更されると、ユーザーオペレーターは Kafka ユーザーに対して対応するアクションを実行します。

アプリケーションのデプロイメントの一部として KafkaUser リソースを宣言すると、User Operator が Kafka ユーザーを管理します。ユーザーの認証および承認メカニズムを指定できます。たとえば、ユーザーがブローカーへのアクセスを独占しないようにするため、Kafka リソースの使用を制御する ユーザークォータ を設定することもできます。

ユーザーが作成されると、ユーザークレデンシャルが Secret に作成されます。アプリケーションはユーザーとそのクレデンシャルを使用して、認証やメッセージの生成または消費を行う必要があります。

User Operator は 認証のクレデンシャルを管理する他に、KafkaUser 宣言にユーザーのアクセス権限の記述を含めることで承認も管理します。

Streams for Apache Kafka Operator はフィーチャーゲートを使用して特定の機能を有効または無効にします。フィーチャーゲートを有効にすると、関連付けられた Operator の動作が変更され、対応する機能が Streams for Apache Kafka デプロイメントに導入されます。

フィーチャーゲートは Operator の設定で指定され、alpha、beta、または Graduated の 3 段階の成熟度があります。成熟フィーチャーゲートは、一般提供 (GA) として永続的に有効化された機能です。

詳細はフィーチャーゲートを参照してください。

CRD をインストールして新規カスタムリソースタイプをクラスターに追加した後に、その仕様に基づいてリソースのインスタンスを作成できます。

Streams for Apache Kafka コンポーネントのカスタムリソースには、spec で定義される共通の設定プロパティーがあります。

Kafka トピックカスタムリソースからのこの抜粋では、apiVersion および kind プロパティーを使用して、関連付けられた CRD を識別します。Spec プロパティーは、トピックのパーティションおよびレプリカの数を定義する設定を示しています。

Kafka トピックカスタムリソース

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: my-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 1
  replicas: 1
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: my-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 1
  replicas: 1
  # ...

共通の設定、特定のコンポーネントに特有の設定など、他にも YAML 定義に組み込むことができる設定オプションが多数あります。

複数のリソースに共通する設定オプションの一部が以下に記載されています。セキュリティー および メトリックコレクション も採用できます (該当する場合)。

共通設定の YAML 例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-cluster
spec:
  # ...
  bootstrapServers: my-cluster-kafka-bootstrap:9092
  resources:
    requests:
      cpu: 12
      memory: 64Gi
    limits:
      cpu: 12
      memory: 64Gi
  logging:
    type: inline
    loggers:
      connect.root.logger.level: INFO
  readinessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  jvmOptions:
    "-Xmx": "2g"
    "-Xms": "2g"
  template:
    pod:
      affinity:
        nodeAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
              - matchExpressions:
                  - key: node-type
                    operator: In
                    values:
                      - fast-network
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-cluster
spec:
  # ...
  bootstrapServers: my-cluster-kafka-bootstrap:9092
  resources:
    requests:
      cpu: 12
      memory: 64Gi
    limits:
      cpu: 12
      memory: 64Gi
  logging:
    type: inline
    loggers:
      connect.root.logger.level: INFO
  readinessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  jvmOptions:
    "-Xmx": "2g"
    "-Xms": "2g"
  template:
    pod:
      affinity:
        nodeAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
              - matchExpressions:
                  - key: node-type
                    operator: In
                    values:
                      - fast-network
  # ...

Kafka クラスターは、1 つまたは複数のブローカーで設定されます。プロデューサーおよびコンシューマーがブローカー内のトピックにアクセスできるようにするには、Kafka 設定でクラスターへのデータの保存方法、およびデータへのアクセス方法を定義する必要があります。ラック全体で複数のブローカーノードを使用して Kafka クラスターを実行するように設定できます。

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      - name: external1
        port: 9094
        type: route
        tls: true
        authentication:
          type: tls
    # ...
    storage:
      type: persistent-claim
      size: 10000Gi
    # ...
    rack:
      topologyKey: topology.kubernetes.io/zone
    config:
      replica.selector.class: org.apache.kafka.common.replica.RackAwareReplicaSelector
    # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      - name: external1
        port: 9094
        type: route
        tls: true
        authentication:
          type: tls
    # ...
    storage:
      type: persistent-claim
      size: 10000Gi
    # ...
    rack:
      topologyKey: topology.kubernetes.io/zone
    config:
      replica.selector.class: org.apache.kafka.common.replica.RackAwareReplicaSelector
    # ...

ノードプールは、Kafka クラスター内の Kafka ノードの個別のグループを指します。ノードプールを使用すると、同じ Kafka クラスター内でノードに異なる設定を持たせることができます。ノードプールで指定されていない設定オプションは、Kafka 設定から継承されます。

1 つ以上のノードプールを含む Kafka クラスターをデプロイできます。ノードプールの設定には、必須の設定とオプションの設定が含まれます。レプリカ、ロール、ストレージの設定は必須です。

KRaft モードを使用している場合は、ノードプール内のすべてのノードがブローカー、コントローラー、またはその両方として動作するようにロールを指定できます。コントローラーと二重のロールは KRaft に固有です。クラスター管理に ZooKeeper を使用する Kafka クラスターを使用している場合は、ブローカーロールのみで設定されたノードプールを使用できます。

ノードプール設定を示す YAML の例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: pool-a
  labels:
    strimzi.io/cluster: my-cluster
spec:
  replicas: 3
  roles:
    - broker
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: pool-a
  labels:
    strimzi.io/cluster: my-cluster
spec:
  replicas: 3
  roles:
    - broker
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false

Kafka MirrorMaker 2 は、データセンター内またはデータセンター全体の 2 台以上の Kafka クラスター間でデータをレプリケーションします。MirrorMaker 2 を設定するには、ソースおよびターゲット (宛先) の Kafka クラスターが実行中である必要があります。

ソースクラスターからターゲットクラスターへのデータの ミラーリング プロセスは非同期です。各 MirrorMaker 2 インスタンスは、1 つのソースクラスターから 1 つのターゲットクラスターにデータをミラーリングします。複数の MirrorMaker 2 インスタンスを使用して、任意の数のクラスター間でデータをミラーリングできます。

図7.1 2 つのクラスターにおけるレプリケーション

MirrorMaker 2 は、ソースおよびターゲットのクラスター設定を次のように使用します。

トピックとコンシューマーグループのレプリケーションは、コンマ区切りのリストまたは正規表現パターンとして指定されます。

MirrorMaker 2 設定を示す YAML の例

  apiVersion: kafka.strimzi.io/v1beta2
  kind: KafkaMirrorMaker2
  metadata:
    name: my-mirror-maker2
    spec:
      version: 3.7.0
      connectCluster: "my-cluster-target"
      clusters:
      - alias: "my-cluster-source"
        bootstrapServers: my-cluster-source-kafka-bootstrap:9092
      - alias: "my-cluster-target"
        bootstrapServers: my-cluster-target-kafka-bootstrap:9092
      mirrors:
      - sourceCluster: "my-cluster-source"
        targetCluster: "my-cluster-target"
        sourceConnector: {}
      topicsPattern: ".*"
      groupsPattern: "group1|group2|group3"

  apiVersion: kafka.strimzi.io/v1beta2
  kind: KafkaMirrorMaker2
  metadata:
    name: my-mirror-maker2
    spec:
      version: 3.7.0
      connectCluster: "my-cluster-target"
      clusters:
      - alias: "my-cluster-source"
        bootstrapServers: my-cluster-source-kafka-bootstrap:9092
      - alias: "my-cluster-target"
        bootstrapServers: my-cluster-target-kafka-bootstrap:9092
      mirrors:
      - sourceCluster: "my-cluster-source"
        targetCluster: "my-cluster-target"
        sourceConnector: {}
      topicsPattern: ".*"
      groupsPattern: "group1|group2|group3"

Kafka MirrorMaker (MirrorMaker 1 とも呼ばれる) は、プロデューサーとコンシューマーを使用して、次のようにクラスター間でデータをレプリケーションします。

コンシューマーとプロデューサの設定には、必要な認証と暗号化の設定が含まれます。include プロパティーは、ソースクラスターからターゲットクラスターにミラーリングするトピックを定義します。

主なコンシューマー設定

キープロデューサーの設定

MirrorMaker 設定の YAML 例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker
metadata:
  name: my-mirror-maker
spec:
  # ...
  consumer:
    bootstrapServers: my-source-cluster-kafka-bootstrap:9092
    groupId: "my-group"
    numStreams: 2
    offsetCommitInterval: 120000
    # ...
  producer:
    # ...
    abortOnSendFailure: false
    # ...
  include: "my-topic|other-topic"
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker
metadata:
  name: my-mirror-maker
spec:
  # ...
  consumer:
    bootstrapServers: my-source-cluster-kafka-bootstrap:9092
    groupId: "my-group"
    numStreams: 2
    offsetCommitInterval: 120000
    # ...
  producer:
    # ...
    abortOnSendFailure: false
    # ...
  include: "my-topic|other-topic"
  # ...

Streams for Apache Kafka の KafkaConnect リソースを使用して、新しい Kafka Connect クラスターをすばやく簡単に作成します。

KafkaConnect リソースを使用して Kafka Connect をデプロイする場合は、Kafka クラスターに接続するためのブートストラップサーバーアドレスを (spec.bootstrapServers で) 指定します。サーバーがダウンした場合に備えて、複数のアドレスを指定できます。また、認証情報と TLS 暗号化証明書を指定して、安全な接続を確立します。

KafkaConnect リソースを使用して、以下を指定することもできます。

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

プラグイン設定

プラグインは、コネクターインスタンスを作成するための実装を提供します。プラグインがインスタンス化されると、特定のタイプの外部データシステムに接続するための設定が提供されます。プラグインは、特定の種類のデータソースに接続するためのコネクターとタスクの実装を定義する 1 つ以上の JAR ファイルのセットを提供します。多くの外部システム用のプラグインは、Kafka Connect で使用できます。独自のプラグインを作成することもできます。

この設定では、Kafka Connect にフィードするソース入力データおよびターゲット出力データを記述します。ソースコネクターの場合、外部ソースデータは、メッセージを格納する特定のトピックを参照する必要があります。プラグインには、データの変換に必要なライブラリーとファイルが含まれている場合もあります。

Kafka Connect デプロイメントには、1 つ以上のプラグインを含めることができますが、各プラグインのバージョンは 1 つだけです。

選択したプラグインを含むカスタム Kafka Connect イメージを作成できます。イメージは次の 2 つの方法で作成できます。

コンテナーイメージを自動的に作成するには、KafkaConnect リソースの build プロパティーを使用して、Kafka Connect クラスターに追加するプラグインを指定します。Streams for Apache Kafka は、プラグインアーティファクトを自動的にダウンロードして新しいコンテナーイメージに追加します。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  # ...
  build: 
    output: 
      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins: 
      - name: my-connector
        artifacts:
          - type: tgz
            url: https://<plugin_download_location>.tgz
            sha512sum: <checksum_to_verify_the_plugin>
      # ...
  # ...

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

    output: 

      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins: 

      - name: my-connector
        artifacts:
          - type: tgz
            url: https://<plugin_download_location>.tgz
            sha512sum: <checksum_to_verify_the_plugin>
      # ...
  # ...

Dockerfile を使用してイメージをビルドしている場合は、Streams for Apache Kafka の最新のコンテナーイメージをベースイメージとして使用して、プラグイン設定ファイルを追加できます。

FROM registry.redhat.io/amq-streams/kafka-37-rhel8:2.7.0
USER root:root
COPY ./my-plugins/ /opt/kafka/plugins/
USER 1001

FROM registry.redhat.io/amq-streams/kafka-37-rhel8:2.7.0
USER root:root
COPY ./my-plugins/ /opt/kafka/plugins/
USER 1001

ワーカー用の Kafka Connect クラスター設定

ワーカーの設定は、KafkaConnect リソースの config プロパティーで指定します。

分散型 Kafka Connect クラスターには、グループ ID と一連の内部設定トピックがあります。

Kafka Connect クラスターは、デフォルトでこれらのプロパティーに同じ値で設定されています。Kafka Connect クラスターは、エラーを作成するため、グループ ID またはトピック名を共有できません。複数の異なる Kafka Connect クラスターを使用する場合、これらの設定は、作成された各 Kafka Connect クラスターのワーカーに対して一意である必要があります。

各 Kafka Connect クラスターで使用されるコネクターの名前も一意である必要があります。

次のワーカー設定の例では、JSON コンバーターが指定されています。レプリケーション係数は、Kafka Connect によって使用される内部 Kafka トピックに設定されます。これは、実稼働環境では少なくとも 3 つ必要です。トピックの作成後にレプリケーション係数を変更しても効果はありません。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
# ...
spec:
  config:
    # ...
    group.id: my-connect-cluster 
    offset.storage.topic: my-connect-cluster-offsets 
    config.storage.topic: my-connect-cluster-configs 
    status.storage.topic: my-connect-cluster-status 
    key.converter: org.apache.kafka.connect.json.JsonConverter 
    value.converter: org.apache.kafka.connect.json.JsonConverter 
    key.converter.schemas.enable: true 
    value.converter.schemas.enable: true 
    config.storage.replication.factor: 3 
    offset.storage.replication.factor: 3 
    status.storage.replication.factor: 3 
  # ...

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

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

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

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

    key.converter: org.apache.kafka.connect.json.JsonConverter 

    value.converter: org.apache.kafka.connect.json.JsonConverter 

    key.converter.schemas.enable: true 

    value.converter.schemas.enable: true 

    config.storage.replication.factor: 3 

    offset.storage.replication.factor: 3 

    status.storage.replication.factor: 3 

  # ...

コネクターの Kafka Connector 管理

デプロイメントでワーカー Pod に使用されるコンテナーイメージにプラグインが追加されたら、Streams for Apache Kafka の KafkaConnector カスタムリソースまたは Kafka Connect API を使用してコネクターインスタンスを管理できます。これらのオプションを使用して、新しいコネクターインスタンスを作成することもできます。

KafkaConnector リソースは、Cluster Operator によるコネクターの管理に OpenShift ネイティブのアプローチを提供します。KafkaConnector リソースを使用してコネクターを管理するには、KafkaConnect カスタムリソースでアノテーションを指定する必要があります。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
  # ...

use-connector-resources を true に設定すると、KafkaConnectors はコネクターを作成、削除、および再設定できます。

KafkaConnect 設定で use-connector-resources が有効になっている場合は、KafkaConnector リソースを使用してコネクターを定義および管理する必要があります。KafkaConnector リソースは、外部システムに接続するように設定されています。これらは、外部データシステムと相互作用する Kafka Connect クラスターおよび Kafka クラスターと同じ OpenShift クラスターにデプロイされます。

設定は、認証を含め、コネクターインスタンスが外部データシステムに接続する方法を指定します。また、監視するデータを指定する必要があります。ソースコネクターの場合は、設定でデータベース名を指定できます。ターゲットトピック名を指定することで、Kafka のどこにデータを配置するかを指定することもできます。

タスクの最大数を指定するには、tasksMax を使用します。たとえば、tasksMax: 2 のソースコネクターは、ソースデータのインポートを 2 つのタスクに分割する場合があります。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-source-connector  
  labels:
    strimzi.io/cluster: my-connect-cluster 
spec:
  class: org.apache.kafka.connect.file.FileStreamSourceConnector 
  tasksMax: 2 
  autoRestart: 
    enabled: true
  config: 
    file: "/opt/kafka/LICENSE" 
    topic: my-topic 
    # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-source-connector  

  labels:
    strimzi.io/cluster: my-connect-cluster 

spec:
  class: org.apache.kafka.connect.file.FileStreamSourceConnector 

  tasksMax: 2 

  autoRestart: 

    enabled: true
  config: 

    file: "/opt/kafka/LICENSE" 

    topic: my-topic 

    # ...

Kafka Connect API

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

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

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"
    }
}'

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"
    }
}'

KafkaConnectors が有効になっている場合、Kafka Connect REST API に直接手作業で追加された変更は Cluster Operator によって元に戻されます。

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

Kafka Bridge 設定には、接続先の Kafka クラスターのブートストラップサーバー仕様と、必須の暗号化および認証オプションが必要になります。

コンシューマーの Apache Kafka 設定ドキュメント および プロデューサーの Apache Kafka 設定ドキュメント で説明されているように、Kafka Bridge コンシューマーおよびプロデューサー設定は標準です。

HTTP 関連の設定オプションでは、サーバーがリッスンするポート接続を設定します。

CORS

Kafka Bridge では、CORS (Cross-Origin Resource Sharing) の使用がサポートされます。CORS は、複数のオリジンから指定のリソースにブラウザーでアクセスできるようにする HTTP メカニズムです (たとえば、異なるドメイン上のリソースへのアクセス)。CORS を使用する場合、Kafka Bridge を通じた Kafka クラスターとの対話用に、許可されるリソースオリジンおよび HTTP メソッドのリストを定義できます。リストは、Kafka Bridge 設定の http 仕様で定義されます。

CORS では、異なるドメイン上のオリジンソース間での シンプルな 要求および プリフライト 要求が可能です。

Kafka ブリッジ設定の YAML 例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaBridge
metadata:
  name: my-bridge
spec:
  # ...
  bootstrapServers: my-cluster-kafka:9092
  http:
    port: 8080
    cors:
      allowedOrigins: "https://strimzi.io"
      allowedMethods: "GET,POST,PUT,DELETE,OPTIONS,PATCH"
  consumer:
    config:
      auto.offset.reset: earliest
  producer:
    config:
      delivery.timeout.ms: 300000
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaBridge
metadata:
  name: my-bridge
spec:
  # ...
  bootstrapServers: my-cluster-kafka:9092
  http:
    port: 8080
    cors:
      allowedOrigins: "https://strimzi.io"
      allowedMethods: "GET,POST,PUT,DELETE,OPTIONS,PATCH"
  consumer:
    config:
      auto.offset.reset: earliest
  producer:
    config:
      delivery.timeout.ms: 300000
  # ...

Streams for Apache Kafka は、暗号化された通信用のプロトコルである Transport Layer Security (TLS)をサポートします。

通信は、以下の間で常に暗号化されます。

Kafka ブローカーとクライアントの間で TLS 暗号化を設定することもできます。Kafka ブローカーの外部リスナーを設定するときに、外部クライアントに対して TLS が指定されます。

Streams for Apache Kafka コンポーネントおよび Kafka クライアントは、暗号化にデジタル署名を使用します。Cluster Operator は、証明書を設定し、Kafka クラスター内で暗号化を有効にします。Kafka クライアントと Kafka ブローカーの通信やクラスター間の通信に、Kafka リスナー証明書と呼ばれる独自のサーバー証明書を指定できます。

Streams for Apache Kafka は Secret を使用して、mTLS に必要な証明書と秘密鍵を PEM および PKCS #12 形式で保存します。

TLS CA (認証局) は、コンポーネントの ID を認証するために証明書を発行します。Streams for Apache Kafka は、CA 証明書に対してコンポーネントの証明書を検証します。

Kafka リスナーは認証を使用して、Kafka クラスターへのクライアント接続のセキュリティーを確保します。

サポート対象の認証メカニズム:

User Operator では mTLS および SCRAM 認証のユーザー認証情報は管理対象ですが、OAuth 2.0 は管理対象ではありません。たとえば、User Operator を使用して、Kafka クラスターにアクセスする必要があるクライアントに対応するユーザーを作成し、認証タイプとして tls を指定できます。

OAuth 2.0 トークンベースの認証を使用すると、アプリケーションクライアントは、アカウント認証情報を公開せずに Kafka ブローカーにアクセスできます。認可サーバーは、アクセスの付与とアクセスに関する問い合わせを処理します。

カスタム認証では、kafka でサポートされているあらゆるタイプの認証が可能です。柔軟性を高めることができますが、複雑さも増します。

Kafka クラスターは、承認メカニズムを使用して特定のクライアントまたはユーザーによって Kafka ブローカーで許可される操作を制御します。承認は、Kafka クラスターに適用されると、クライアント接続に使用する全リスナーに対して有効になります。

ユーザーを Kafka ブローカー設定の スーパーユーザー のリストに追加すると、承認メカニズムにより適用される承認制約に関係なく、そのユーザーにはクラスターへのアクセスが無制限に許可されます。

サポート対象の承認メカニズム:

簡易認可では、AclAuthorizer および StandardAuthorizer Kafka プラグインを使用し、さまざまなリソースへのユーザーアクセスを指定するアクセスコントロールリスト (ACL) を管理します。カスタム認可の場合は、ACL ルールを適用するように独自の Authorizer プラグインを設定します。

OAuth 2.0 および OPA は、認可サーバーからのポリシーベースの制御を提供します。Kafka ブローカーのリソースへのアクセス権限を付与するのに使用されるセキュリティーポリシーおよびパーミッションは、認可サーバーで定義されます。

認可サーバーに接続し、クライアントまたはユーザーが要求した操作が許可または拒否されることを検証するのに、URL が使用されます。ユーザーとクライアントは、認可サーバーで作成されるポリシーと照合され、Kafka ブローカーで特定のアクションを実行するためのアクセスが許可されます。

連邦情報処理標準 (FIPS) は、情報システムによって処理または送信される機密データおよび情報の機密性、完全性、および可用性を確保するために米国政府によって確立された一連のセキュリティー標準です。Streams for Apache Kafka コンテナーイメージで使用される OpenJDK は、FIPS 対応の OpenShift クラスター上で実行すると、FIPS モードを自動的に有効にします。

NIST 検証プログラムおよび検証済みモジュールの詳細は、NIST Web サイトの Cryptographic Module Validation Program を参照してください。

Prometheus は、Kafka コンポーネントおよび Streams for Apache Kafka Operator からメトリクスデータを抽出できます。

Prometheus を使用してメトリックデータを取得し、アラートを発行するには、Prometheus および Prometheus Alertmanager プラグインをデプロイする必要があります。メトリックデータを公開するには、Kafka リソースもメトリック設定でデプロイまたは再デプロイする必要があります。

Prometheus は、公開されたメトリクスデータをモニタリング用に収集します。Alertmanager は、事前に定義されたアラートルールをもとに、条件が問題発生の可能性を示した場合に、アラートを発行します。

メトリクスおよびアラートルール設定ファイルのサンプルは、Streams for Apache Kafka に同梱されています。Streams for Apache Kafka で提供されるアラートメカニズムのサンプルは、通知を Slack チャネルに送信するように設定されています。

Grafana は Prometheus によって公開されるメトリクスデータを使用して、モニタリングできるように、ダッシュボードを視覚化して表示します。

データソースとして Prometheus を追加している場合には、Grafana のデプロイメントが必要です。ダッシュボードの例(Streams for Apache Kafka が JSON ファイルとして提供される)は、モニタリングデータを表示するために、Grafana インターフェイスを使用してインポートされます。

Kafka Exporter は、Apache Kafka ブローカーおよびクライアントのモニタリングを強化するオープンソースプロジェクトです。Kafka Exporter は、Kafka クラスターとともにデプロイされ、オフセット、コンシューマーグループ、コンシューマーラグ、およびトピックに関連する Kafka ブローカーからの Prometheus メトリックデータを追加で抽出します。提供される Grafana ダッシュボードを使用して、Prometheus が Kafka Exporter から収集したデータを可視化することができます。

サンプル設定ファイル、アラートルール、および Kafka Exporter の Grafana ダッシュボードは、Streams for Apache Kafka で提供されます。

分散トレースは、分散システム内のアプリケーション間のトランザクションの進行状況を追跡します。マイクロサービスのアーキテクチャーでは、トレースがサービス間のトランザクションの進捗を追跡します。トレースデータは、アプリケーションのパフォーマンスを監視し、ターゲットシステムおよびエンドユーザーアプリケーションの問題を調べるのに有用です。

Streams for Apache Kafka では、トレースにより、ソースシステムから Kafka、さらに Kafka からターゲットシステムおよびアプリケーションへのメッセージのエンドツーエンドの追跡が容易になります。分散トレースは、Grafana ダッシュボードおよびコンポーネントロガーでのメトリックの監視を補完します。

トレースのサポートは、以下の Kafka コンポーネントに組み込まれています。

トレースは Kafka ブローカーではサポートされません。

カスタムリソースを使用して、これらのコンポーネントのトレースを有効にして設定します。spec.template プロパティーを使用してトレース設定を追加します。

spec.tracing.type プロパティーを使用してトレースタイプを指定することにより、トレースを有効にします。

Kafka クライアントのトレース

Kafka プロデューサーやコンシューマーなどのクライアントアプリケーションも、トランザクションをモニタリングするように設定できます。クライアントはトレースプロファイルで設定され、トレーサーはクライアントアプリケーションが使用するように初期化されます。

Cruise Control は、次の Kafka 操作をサポートするオープンソースシステムです。

この操作は、ブローカー Pod をより効率的に使用する、よりバランスの取れた Kafka クラスターを実行するのに役立ちます。

通常、クラスターの負荷は時間とともに不均等になります。大量のメッセージトラフィックを処理するパーティションは、使用可能なブローカー全体で均等に分散されない可能性があります。クラスターを再分散するには、管理者はブローカーの負荷を監視し、トラフィックの多いパーティションを容量に余裕のあるブローカーに手作業で再割り当てします。

Cruise Control はクラスターのリバランス処理を自動化します。CPU、ディスク、およびネットワーク負荷を基にして、クラスターにおけるリソース使用のワークロードモデルを構築し、パーティションの割り当てをより均等にする、最適化プロポーザル (承認または拒否可能) を生成します。これらのプロポーザルの算出には、設定可能な最適化ゴールが複数使用されます。

特定のモードで最適化の提案を生成できます。デフォルトの full モードでは、すべてのブローカー間でパーティションがリバランスされます。add-brokers および remove-brokers モードを使用して、クラスターをスケールアップまたはスケールダウンするときの変更に対応することもできます。

最適化プロポーザルを承認すると、Cruise Control はそのプロポーザルを Kafka クラスターに適用します。KafkaRebalance リソースを使用して、最適化の提案を設定および生成します。最適化の提案が自動または手動で承認されるように、アノテーションを使用してリソースを設定できます。