AMQ Streams 1.7 は OpenShift Container Platform 4.6 および 4.7 でサポートされます。

サポートされるプラットフォームバージョンの詳細については、Red Hat ナレッジベースの記事「Red Hat AMQ 7 でサポートされる構成」を参照してください。

AMQ Streams は Apache Kafka バージョン 2.7.0 に対応するようになりました。

AMQ Streams は Kafka 2.7.0 を使用します。Red Hat によってビルドされた Kafka ディストリビューションのみがサポートされます。

ブローカーおよびクライアントアプリケーションを Kafka 2.7.0 にアップグレードする前に、Cluster Operator を AMQ Streams バージョン 1.7 にアップグレードする必要があります。アップグレードの手順は、「AMQ Streams のアップグレード」を参照してください。

詳細は、Kafka 2.6.0 および Kafka 2.7.0 のリリースノートを参照してください。

サポートされるバージョンの詳細は、カスタマーポータルの「Red Hat AMQ 7 コンポーネントの詳細」を参照してください。

Kafka 2.7.0 には、Kafka 2.6.x と同じ ZooKeeper バージョン (ZooKeeper バージョン 3.5.8) が必要です。そのため、AMQ Streams 1.6 から AMQ Streams 1.7 にアップグレードするときに、Cluster Operator は ZooKeeper のアップグレードを実行しません。

AMQ Streams 1.7 では、AMQ Streams カスタムリソースのスキーマを更新する v1beta2 API バージョンが導入されました。古い API バージョンは非推奨になりました。

AMQ Streams 1.7 にアップグレードした後、API バージョン v1beta2 を使用するようにカスタムリソースをアップグレードする 必要 があります。これは、アップグレード後にいつでも実行できますが、AMQ Stremas の次回のマイナーバージョン更新 (AMQ Streams 1.8) までに完了する必要があります。

カスタムリソースのアップグレードをサポートするため、Red Hat AMQ Streams 1.7.0 API 変換ツール が提供されます。AMQ Streams のダウンロードサイト から API 変換ツールをダウンロードできます。ツールの使用方法は、ドキュメントおよび提供される readme に記載されています。

カスタムリソースを v1beta2 にアップグレードすると、Kubernetes 1.22 に必要な Kubernetes CRD v1 が AMQ Streams によって準備されます。

1 回のアップグレードで、以前の AMQ Streams バージョンから直接最新の AMQ Streams バージョンにアップグレードできるようになりました。たとえば、中間のバージョンを省略して、AMQ Streams 1.5 から直接 AMQ Streams 1.7 にアップグレードできます。

「AMQ Streams のアップグレード」を参照してください。

AMQ Streams がデータコネクションに必要なコネクタープラグインでコンテナーイメージを自動的に ビルド するために、ビルド設定を使用できるようになりました。

AMQ Streams で新しいイメージを自動的に作成するには、ビルド 設定にはコンテナーイメージを保存するコンテナーレジストリーを参照する 出力 プロパティーと、イメージに追加するコネクター プラグイン とそれらのアーティファクトをリストするプラグインプロパティーが必要です。

ビルド 設定はイメージストリームを参照することもできます。イメージストリームは、OpenShift Container Platform の統合レジストリーに保存されているコンテナーイメージを参照します。

出力 プロパティーは、イメージのタイプおよび名前を記述し、任意でコンテナーレジストリーへのアクセスに必要なクレデンシャルが含まれる Secret の名前を記述します。plugins プロパティーは、アーティファクトのタイプとアーティファクトのダウンロード元となる URL を記述します。さらに、SHA-512 チェックサムを指定して、アーティファクトを展開する前に検証することもできます。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
spec:
  # ...
  build:
    output:
      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins:
      - name: debezium-postgres-connector
        artifacts:
          - type: tgz
            url: https://ARTIFACT-ADDRESS.tgz
            sha512sum: HASH-NUMBER-TO-VERIFY-ARTIFACT
      # ...
  #...

以下を参照してください。

メトリクスは、カスタムリソースのデプロイ時に自動作成される ConfigMap を使用して設定されるようになりました。

metricsConfig プロパティーを使用して、Kafka コンポーネントの Prometheus メトリクスを有効化および設定します。metricsConfig プロパティーには、Prometheus JMX exporter の追加設定が含まれる ConfigMap への参照が含まれます。

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: my-config-map
          key: my-key
    # ...
  zookeeper:
    # ...

ConfigMap は、JMX Prometheus エクスポーターの YAML 設定をキーの下に保存します。

kind: ConfigMap
apiVersion: v1
metadata:
  name: my-configmap
data:
  my-key: |
    lowercaseOutputName: true
    rules:
    # Special cases and very specific rules
    - pattern: kafka.server<type=(.+), name=(.+), clientId=(.+), topic=(.+), partition=(.*)><>Value
      name: kafka_server_$1_$2
      type: GAUGE
      labels:
       clientId: "$3"
       topic: "$4"
       partition: "$5"
    # further configuration

追加設定なしで Prometheus メトリクスのエクスポートを有効にするには、metricsConfig.valueFrom.configMapKeyRef.key 配下に空のファイルが含まれる ConfigMap を参照します。空のファイルを参照する場合、名前が変更されていない限り、すべてのメトリクスが公開されます。

4章非推奨の機能 の説明にあるように、spec.metrics プロパティーは非推奨になりました。

「共通の設定プロパティー」を参照してください。

Red Hat Debezium は分散型の変更データキャプチャープラットフォームです。データベースの行レベルの変更をキャプチャーして、変更イベントレコードを作成し、Kafka トピックにレコードをストリーミングします。Debezium は Apache Kafka に構築されます。AMQ Streams で Debezium をデプロイおよび統合できます。AMQ Streams のデプロイ後に、Kafka Connect で Debezium をコネクター設定としてデプロイします。Debezium は変更イベントレコードを OpenShift 上の AMQ Streams に渡します。アプリケーションは 変更イベントストリーム を読み取りでき、変更イベントが発生した順にアクセスできます。

Debezium には、以下を含む複数の用途があります。

Debezium は、以下の共通データベースのコネクター (Kafka Connect をベースとする) を提供します。

AMQ Streams での Debezium のデプロイについて、詳しくは 製品ドキュメント を参照してください。

Service Registry は、データストリーミングのサービススキーマの集中型ストアとして使用できます。Kafka では、Service Registry を使用して Apache Avro または JSON スキーマを格納できます。

Service Registry は、REST API および Java REST クライアントを提供し、サーバー側のエンドポイントを介してクライアントアプリケーションからスキーマを登録およびクエリーします。

Service Registry を使用すると、クライアントアプリケーションの設定からスキーマ管理のプロセスが分離されます。クライアントコードに URL を指定して、アプリケーションがレジストリーからスキーマを使用できるようにします。

たとえば、メッセージをシリアライズおよびデシリアライズするスキーマをレジストリーに保存できます。アプリケーションは保存されたスキーマを参照し、それらを使用して送受信するメッセージとスキーマとの互換性を維持します。

Kafka クライアントアプリケーションは実行時にスキーマを Service Registry からプッシュまたはプルできます。

AMQ Streams で Service Registry を使用するための詳細は、「製品ドキュメント」を参照してください。

Kafka 2.7.0 に導入された改良機能の概要は『Kafka 2.7.0 Release Notes』を参照してください。

Kafka Connect、MirrorMaker、および Kafka Bridge の Deployment ストラテジーを設定できるようになりました。

RollingUpdate ストラテジーはデフォルトですべてのリソースに使用されます。Kafka クラスターのローリングアップデート中に、Deployment の古い Pod と新しい Pod が並行して実行されます。これは、ほとんどのユースケースに最適なストラテジーです。

リソースの消費を減らすには、再作成 ストラテジーを選択します。このストラテジーでは、ローリングアップデート中に、Deployment の古い Pod は新規 Pod が作成される前に終了します。

KafkaConnect、KafkaMirrorMaker、KafkaMirrorMaker2、および KafkaBridge リソースの spec.template.deployment で Deployment ストラテジーを設定します。

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnect
metadata:
  name: my-connect-cluster
spec:
  #...
  template:
    deployment:
      deploymentStrategy: Recreate
  #...

spec.template.deployment が設定されていない場合は、RollingUpdate ストラテジーが使用されます。

「 DeploymentTemplate スキーマ参照」を参照してください。

クラスターおよびクライアント CA Secret は、Kafka カスタムリソースに設定される ownerReference フィールドで作成されます。

generateSecretOwnerReference: false プロパティーを Kafka クラスター設定に追加することで、CA Secret ownerReference を無効にすることができます。CA Secret の ownerReference が無効になっている場合、対応する Kafka カスタムリソースが削除されると Secret は OpenShift によって削除されません。その後、CA Secret は新しい Kafka クラスターで再利用できます。

apiVersion: kafka.strimzi.io/v1beta1
kind: Kafka
# ...
spec:
# ...
  clusterCa:
    generateCertificateAuthority: true
    generateSecretOwnerReference: false
  clientsCa:
    generateCertificateAuthority: true
    generateSecretOwnerReference: false
# ...

「 CA Secret での ownerReference の無効化」を参照してください。

secretPrefix プロパティーを使用して、KafkaUser リソース用に作成されたすべてのシークレット名にプレフィックスを追加する User Operator を設定できるようになりました。

例として、以下の設定を見てみましょう。

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
  zookeeper:
    # ...
  entityOperator:
    # ...
    userOperator:
      secretPrefix: kafka-
    # ...

my-user という名前のユーザーに、kafka-my-user という名前のシークレットを作成します。

「 EntityUserOperatorSpec スキーマ参照」を参照してください。

アノテーションを使用すると、Kafka クラスターまたは ZooKeeper クラスター StatefulSets の一部である既存 Pod のローリングアップデートを手動でトリガーできます。同じ StatefulSet の複数の Pod に同時にアノテーションが付けられると、連続したローリングアップデートは同じ調整実行内で実行されます。

「Pod アノテーションを使用したローリングアップデートの実行」を参照してください。

AMQ Streams では、ZooKeeper を使用してトピックメタデータを保存しなくなりました。Karaf メタデータは Kafka クラスターに格納され、Topic Operator の制御下に置かれるようになりました。

この変更は、将来的に ZooKeeper が Kafka の依存関係でなくなることを想定して、AMQ Streams を準備するために必要です。

Topic Operator は永続ストレージを使用して、トピック設定をキーと値のペアとして記述するトピックメタデータを保存するようになりました。トピックメタデータは、ローカルのインメモリーでアクセスされます。ローカルのインメモリートピックストアに適用される操作からの更新は、ディスク上のバックアップトピックストアに永続化されます。トピックストアは、Kafka トピックからの更新と継続的に同期されます。

AMQ Streams 1.7 にアップグレードする場合、Topic Operator によってトピックストアが制御されるようにシームレスに移行されます。メタデータは ZooKeeper から検出および移行され、古いストアから削除されます。

「Topic Operator のトピックストア」を参照してください。

sasl.jaas.config プロパティーの JAAS 設定文字列が、SCRAM-SHA-512 認証のある KafkaUser の生成されたシークレットに追加されました。

「SCRAM-SHA-512 認証」を参照してください。

KafkaStatus スキーマは、Kafka クラスターの識別のために clusterId が含まれるように更新されます。Kafka リソースの status プロパティーによって、Kafka クラスターのステータス情報を提供します。

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
  # ...
status:
  conditions:
    lastTransitionTime: "YEAR-MONTH-20T11:37:00.706Z"
    status: "True"
    type: Ready
  observedGeneration: 1
  clusterId: CLUSTER-ID
  # ...

Kafka リソースのステータスを取得すると、Kafka クラスターの ID も返されます。

oc get kafka MY-KAFKA-CLUSTER -o jsonpath='{.status}'

Kafka リソースのクラスター ID のみを取得することもできます。

oc get kafka MY-KAFKA-CLUSTER -o jsonpath='{.status.clusterId}'

「KafkaStatus スキーマ参照」 および「 カスタムリソースのステータスの検索」を参照してください。

KafkaConnector リソースのステータスを取得すると、コネクターによって使用されるトピックのリストが topics プロパティーで返されるようになりました。

「 KafkaConnectorStatus スキーマ参照 」および「 カスタムリソースのステータスの検索」を参照してください。

読み取り専用のルートファイルシステムで AMQ Streams を実行できるようになりました。一時ファイルがマウントされた /tmp ファイルに書き込まれるように、追加のボリュームが追加されました。以前は、/tmp ディレクトリーがコンテナーから直接使用されていました。

このやり方では、コンテナーファイルシステムを変更する必要はなく、AMQ Streams を読み取り専用のルートファイルシステムからスムーズに実行することができます。

AMQ Streams で提供される Kafka 設定ファイルのサンプルで inter.broker.protocol.version が指定されるようになりました。Kafka 設定 の inter.broker.protocol.version および log.message.format.version プロパティーは、指定された Kafka バージョン(spec.kafka.version)によってサポートされるバージョンです。プロパティーは、メッセージに追加されるログ形式のバージョンと、Kafka クラスターで使用されるプロトコルのバージョンを表します。Kafka バージョンのアップグレード時に、これらのプロパティーの更新が必要になります。

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    version: 2.7.0
    #...
    config:
      #...
      log.message.format.version: 2.7
      inter.broker.protocol.version: 2.7

「Kafka のアップグレード」を参照してください。

Cluster Operator は、管理するリソースと同じ namespace または別の namespace で実行できます。2 つの新しい環境変数によって、Cluster Operator にアクセスできる namespace が制御されるようになりました。

デフォルトでは、STRIMZI_OPERATOR_NAMESPACE 環境変数が Kubernetes Downward API を使用して Cluster Operator が稼働している namespace を検索するように設定されます。Cluster Operator がリソースと同じ namespace で実行されている場合は、ローカルアクセスのみが必要で、Strimzi によって許可されます。

Cluster Operator が管理するリソースとは別の namespace で実行されている場合、ネットワークポリシーが設定されている場合を除き、Kubernetes クラスターのすべての namespace は Cluster Operator へのアクセスが許可されます。オプションの STRIMZI_OPERATOR_NAMESPACE_LABELS 環境変数を使用して、namespace ラベルを使用して Cluster Operator のネットワークポリシーを確立します。namespace ラベルを追加すると、Cluster Operator へのアクセスは指定された namespace に限定されます。

#...
env:
  - name: STRIMZI_OPERATOR_NAMESPACE_LABELS
    value: label1=value1,label2=value2
  #...

「Cluster Operator の設定」を参照してください。

KafkaカスタムリソースでclusterCaCertテンプレートプロパティを構成することで、クラスタオペレータが作成したクラスタCAシークレットにカスタムラベルやアノテーションを追加することができます。ラベルとアノテーションは、オブジェクトを特定し、コンテキスト情報を追加するのに便利です。Strimzi カスタムリソースでテンプレートプロパティーを設定します。

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    template:
      clusterCaCert:
        metadata:
          labels:
            label1: value1
            label2: value2
          annotations:
            annotation1: value1
            annotation2: value2
    # ...

「OpenShift リソースのカスタマイズ」を参照してください。

設定で strimzi.io/pause-reconliation アノテーションを true に設定すると、カスタムリソースの調整 を一時停止できます。たとえば、Cluster Operator による調整が一時停止されるように、アノテーションを KafkaConnect リソースに適用できます。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  annotations:
    strimzi.io/pause-reconciliation: "true"
    strimzi.io/use-connector-resources: "true"
  creationTimestamp: 2021-03-12T10:47:11Z
  #...
spec:
  # ...
status:
  conditions:
  - lastTransitionTime: 2021-03-12T10:47:41.689249Z
    status: "True"
    type: ReconciliationPaused

「カスタムリソースの調整の一時停止」を参照してください。

関連するカスタムリソースで Kubernetes アノテーションを使用して、コネクターインスタンスとそのタスクを再起動できるようになりました。

Kafka Connect と Mirror Maker 2.0 の両方のコネクターを再起動できます。MirrorMaker 2.0 は、Kafka Connect フレームワークを使用して、ソースとターゲットの Kafka クラスター間でデータをレプリケートします。

アノテーションを使用して、コネクターの指定タスクを再起動することもできます。

Kafka Connect の場合は、「Kafka コネクターの再起動の実行」と「Kafka コネクタータスクの再起動の実行」を参照してください。

MirrorMaker 2.0 の場合は、「Kafka MirrorMaker 2.0 コネクターの再起動の実行」 と「Kafka MirrorMaker 2.0 コネクタータスクの再起動の実行」を参照してください。

本リリースには、AMQ Streams で OAuth 2.0 トークンベースの認証および承認に対して改良された以下の機能が含まれています。

JWT アクセストークンのチェック

JWT アクセストークンに、2 つの追加チェックを設定できるようになりました。これらのチェックはいずれも Kafka ブローカーリスナーの OAuth 2.0 設定で設定されます。

アクセストークンに必要なデータが含まれていないと拒否されます。イントロスペクションエンドポイントトークン検証を使用する場合は、カスタムチェックがイントロスペクションエンドポイントの応答 JSON に適用されます。

カスタムクレームチェックを設定するには、customClaimCheck オプションを追加して JsonPath フィルタークエリーを定義します。カスタムクレームチェックはデフォルトで無効になっています。

「Kafka ブローカーの OAuth 2.0 サポートの設定」を参照してください。

オーディエンスチェックが有効な場合、Kafka ブローカーは aud クレームにブローカーの clientId が含まれていないトークンを拒否します。

オーディエンスチェックを有効にするには、checkAudience オプションを true に設定します。オーディエンスチェックはデフォルトで無効になっています。

「Kafka ブローカーの OAuth 2.0 サポートの設定」を参照してください。

SASL PLAIN 認証での OAuth 2.0 のサポート

Kafka クライアントと Kafka ブローカー間の OAuth 2.0 認証に PLAIN メカニズムを設定できるようになりました。これまでは、認証されるメカニズムは OAUTHBEARER のみでした。

PLAIN は、すべての Kafka クライアントツール (kafkacat などの開発者ツールを含む) によって使用される簡易認証メカニズムです。AMQ Streams には、PLAIN を OAuth 2.0 認証と使用できるようにするサーバー側のコールバックが含まれています。これらの機能は OAuth 2.0 over PLAIN と呼ばれます。

提供される OAuth 2.0 over PLAIN コールバックと併用すると、Kafka クライアントは以下のいずれかの方法を使用して Kafka ブローカーで認証することができます。

PLAIN を使用するには、Kafka ブローカーの oauth リスナー設定で有効にする必要があります。新しい 3 つの設定オプションがサポートされるようになりました。

  # ...
  name: external
  port: 9094
  type: loadbalancer
  tls: true
  authentication:
    type: oauth
    # ...
    checkIssuer: false
    fallbackUserNameClaim: client_id
    fallbackUserNamePrefix: client-account-
    validTokenType: bearer
    userInfoEndpointUri: https://OAUTH-SERVER-ADDRESS/auth/realms/external/protocol/openid-connect/userinfo
    enableOauthBearer: false 1
    enablePlain: true 2
    tokenEndpointUri: https://OAUTH-SERVER-ADDRESS/auth/realms/external/protocol/openid-connect/token 3
    #...

「OAuth 2.0 認証メカニズム」および「Kafka ブローカーの OAuth 2.0 サポートの設定」を参照してください。

Kafka Connect で新しい rack プロパティーが利用できるようになりました。ラックアウェアネス (Rack awareness) は、異なるラック全体でレプリカを分散するために設定されます。Kafka Connect クラスターの ラック を設定すると、コンシューマーは最も近いレプリカからデータを取得できます。これは、Kafka クラスターが複数のデータセンターにまたがる場合に便利です。

topology キーはクラスターノードのラベルと一致する必要があります。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
#...
spec:
  #...
  rack:
    topologyKey: topology.kubernetes.io/zone

「 KafkaConnectSpec スキーマ参照 」および「 KafkaConnectS2ISpec スキーマ参照」を参照してください。

Pod トポロジー分散制約 は、以下の AMQ Streams カスタムリソースでサポートされるようになりました。

Pod トポロジー分散制約により、Kafka 関連の Pod をノード、ゾーン、リージョン、またはその他のユーザー定義のドメインに分散できます。これらは、Pod スケジューリングの既存の アフィニティー および 容認 プロパティーと共に使用できます。

制約は、関連するカスタムリソースの template.pod.topologySpreadConstraints プロパティーに指定されます。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
#...
spec:
  # ...
  template:
    pod:
      topologySpreadConstraints:
      - maxSkew: "1"
        whenUnsatisfiable: DoNotSchedule
        labelSelector:
          matchLabels:
            label1: value1
#...

参照:

Cruise Control をデプロイし、これを使用して 最適化ゴール (CPU、ディスク、ネットワーク負荷などに定義された制約) を使用し、Kafka をリバランスできます。バランス調整された Kafka クラスターでは、ワークロードがブローカー Pod 全体に均等に分散されます。

Cruise Control は Kafka リソースの一部として設定され、デプロイされます。デフォルトの最適化ゴールを使用するか、要件に合わせて変更できます。Cruise Control の YAML 設定ファイルのサンプルは、examples/cruise-control/ にあります。

Cruise Control がデプロイされると、KafkaRebalance カスタムリソースを作成して以下を行うことができます。

異常検出、通知、独自ゴールの作成、トピックレプリケーション係数の変更などの、その他の Cruise Control の機能は現在サポートされていません。

「Cruise Control によるクラスターのリバランス」を参照してください。

AMQ Streams 1.7 では、1章機能 の説明に従って、KafkaConnect リソースに ビルド 設定が導入されています。build 設定が KafkaConnect リソースに導入されたため、AMQ Streams はデータコネクションに必要なコネクタープラグインでコンテナーイメージを自動的にビルドできるようになりました。

そのため、S2I (Source-to-Image) 対応の Kafka Connect のサポートが非推奨になりました。

この変更に備えるため、Kafka Connect S2I インスタンスを Kafka Connect インスタンスに移行できます。

「Kafka Connect S2I の Kafka Connect への移行」を参照してください。

メトリクス設定は、Kafka コンポーネントの ConfigMap として指定されるようになりました。以前のバージョンでは、spec.metrics プロパティーが使用されていました。

設定を更新し、Prometheus メトリクスのエクスポートを有効にするには、.spec.metrics プロパティーの設定に一致する新しい ConfigMap を作成する必要があります。.spec.metricsConfig プロパティーは、1章機能 の説明に従って ConfigMap を指定するために使用されます。

「AMQ Streams のアップグレード」を参照してください。

v1beta2 が導入され、カスタムリソースのスキーマが更新されるようになりました。古い API バージョンは非推奨になりました。

以下の AMQ Streams カスタムリソースでは、v1alpha1 API バージョンは非推奨になりました。

v1beta1 API バージョンは、以下の AMQ Streams カスタムリソースで非推奨となりました。

「AMQ Streams カスタムリソースのアップグレード」を参照してください。

以下のアノテーションは非推奨となり、AMQ Streams 1.8.0 で削除されます。

AMQ Streams Cluster Operator は、IPv6 (Internet Protocol version 6) クラスターでは起動しません。

この問題を回避する方法は 2 つあります。

「Kafka Bridge 向けの 3scale のデプロイ」で説明されているように、Red Hat 3scale は Kafka Bridge サービスを 検出できません。