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

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

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

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

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

Incremental Cooperative Rebalancing を利用するには、コンシューマーおよび Kafka Streams アプリケーションをアップグレードして、Eager Rebalance Protocol の代わりに新しいプロトコルを使用する必要があります。

「コンシューマーおよび Kafka Streams アプリケーションの Cooperative Rebalancing へのアップグレード」および Apache Kafka ドキュメントの「Notable changes in 2.4.0」を参照してください。

AMQ Streams は、KafkaConnector という新しいカスタムリソースと内部 Operator を使用して、Kafka Connect クラスターでコネクターの Kubernetes ネイティブな管理を提供するようになりました。

KafkaConnector YAML ファイルには、コネクターインスタンスの新規作成または稼働中のコネクターインスタンスの管理を行うために Kubernetes クラスターにデプロイするソースまたはシンクコネクターの設定が記述されます。他の Kafka リソースと同様に、稼働中のコネクターインスタンスは KafkaConnectors に定義された設定と一致するように Cluster Operator によって更新されます。

Installation and Example Files の examples/connector/source-connector.yaml に KafkaConnector リソースの例が含まれるようになりました。YAML ファイルの例をデプロイし、my-topic というトピックでメッセージとしてライセンスファイルの各行を Kafka に送信する FileStreamSourceConnector を作成します。

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

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

以下のタイプのリスナーに、独自のサーバー証明書と秘密鍵を提供できるようになりました。

これらユーザー提供の証明書は Kafka リスナー証明書 と呼ばれます。

組織のプライベート認証局 (CA) またはパブリック CA を使用して、独自の Kafka リスナー証明書を生成および署名できます。

# ...
listeners:
  plain: {}
  external:
    type: loadbalancer
    configuration:
      brokerCertChainAndKey:
        secretName: my-secret
        certificate: my-listener-certificate.crt
        key: my-listener-key.key
    tls: true
    authentication:
      type: tls
# ...

「Kafka リスナー証明書」および「独自のKafka リスナー証明書の指定」を参照してください。

OAuth 2.0 認証のサポートは、テクノロジープレビューから AMQ Streams の一般利用可能なコンポーネントに移行されます。

AMQ Streams は、SASL OAUTHBEARER メカニズムを使用して OAuth 2.0 認証の使用をサポートします。OAuth 2.0 のトークンベースの認証を使用すると、アプリケーションクライアントはアカウントのクレデンシャルを公開せずにアプリケーションサーバー (「リソースサーバー」と呼ばれる) のリソースにアクセスできます。クライアントは、認証手段としてアクセストークンを提示します。アプリケーションサーバーはこのアクセストークンを使用して、付与されたアクセス権限のレベルに関する情報も検索できます。承認サーバーは、アクセスの付与とアクセスに関する問い合わせを処理します。

AMQ Streams のコンテキストでは以下が行われます。

ブローカーおよびクライアントは、必要に応じて OAuth 2.0 承認サーバーと通信し、アクセストークンを取得または検証します。

AMQ Streams のデプロイメントでは、OAuth 2.0 インテグレーションは以下を提供します。

Red Hat Red Hat Single Sign-On を使用して、以下を実行できます。

「OAuth 2.0 トークンベース認証の使用」を参照してください。

Dabezium for Change Data Capture は、データベースを監視し、変更イベントストリームを作成する分散プラットフォームです。Debezium は Apache Karaf に構築され、AMQ Streams とデプロイおよび統合できます。AMQ Streams のデプロイ後に、Kafka Connect で Debezium をコネクター設定としてデプロイします。Debezium によって、データベーステーブルの行レベルの変更がキャプチャーされ、対応する変更イベントが AMQ Streams on OpenShift に渡されます。アプリケーションは 変更イベントストリーム を読み取りでき、変更イベントが発生した順にアクセスできます。

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

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

AMQ Streams で Debezium をデプロイするための詳細は、「製品ドキュメント」を参照してください。

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

AMQ Streams on OpenShift の Kafka Bridge コンポーネントで、Jaeger を使用した分散トレースがサポートされるようになりました。

Kafka Bridge は、Kafka Bridge と HTTP クライアントとの間でメッセージを送受信する場合や、HTTP クライアントがリクエストを Kafka Bridge REST API に送信しコンシューマーを作成してメッセージを取得する場合などに、トレースを生成します。これらのトレースは Jaeger ユーザーインターフェースで表示できます。

Kafka Bridge のトレースを有効にするには、Jaeger トレースの KafkaBridge カスタムリソースを設定します。以下に例を示します。

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaBridge
metadata:
  name: my-bridge
spec:
  #...
  template:
    bridgeContainer:
      env:
        - name: JAEGER_SERVICE_NAME
          value: my-jaeger-service
        - name: JAEGER_AGENT_HOST
          value: jaeger-agent-name
        - name: JAEGER_AGENT_PORT
          value: "6831"
  tracing:
    type: jaeger
#...

kubectl apply を使用して、Kafka クラスターのリソースを更新します。リソースが更新されると、設定に基づいた Jaeger トレーサーが Kafka Bridge によって初期化されます。

「分散トレーシング」および「MirrorMaker、Kafka Connect、および Kafka Bridge リソースでのトレースの有効化」を参照してください。

ユーザークォータは、Kafka ブローカーへのアクセスに関し、ユーザーが定義されたアクセスレベルを超えないようにします。KafkaUser リソースで 2 種類のユーザークォータを設定できるようになりました。

ユーザークォータを設定するには、KafkaUser.spec.quotas リソースの KafkaUser プロパティーを編集します。

「Kafka User リソース」、「KafkaUser スキーマ参照」、および Apache Kafka ドキュメントの「Quotas」を参照してください。

AMQ Streams は Secrets を使用して、Kafka クラスターコンポーネントおよびクライアントの秘密鍵および証明書を格納します。Secrets は、Kafka ブローカー間およびブローカーとクライアント間で TLS で暗号化された接続を確立するために使用されます。Secret は相互 TLS 認証にも使用されます。

PKCS #12 は、暗号化オブジェクトをパスワードで保護された単一のファイルに格納するためのアーカイブファイル形式 (.p12) を定義します。PKCS #12 を使用して、証明書および鍵を 1 カ所で管理できるようになりました

「PKCS #12 ストレージ」を参照してください。

Kafka Connect ベースイメージから Docker イメージを作成するときに Dockerfile に指定された USER が変更になりました。

「Kafka Connect ベースイメージからの Docker イメージの作成」を参照してください。

トークンベースの認証に OAuth 2.0 を使用している場合、Keycloak を使用して、クライアントの Kafka ブローカーへのアクセスを制限する承認ルールを設定できるようになりました。

OAuth 2.0 トークンベースの承認であるこのテクノロジープレビューは、Red Hat Single Sign-On 7.3 ではサポートされません。この機能を試す場合は、Keycloak 8.0.2 を承認サーバーとする開発環境での使用はテストされています。

AMQ Streams は、Keycloak の Authorization Services による OAuth 2.0 トークンベースの承認をサポートします。これにより、セキュリティーポリシーとパーミッションの一元的な管理が可能になります。

Keycloak で定義されたセキュリティーポリシーおよびパーミッションは、Kafka ブローカーのリソースへのアクセスを付与するために使用されます。ユーザーとクライアントは、Kafka ブローカーで特定のアクションを実行するためのアクセスを許可するポリシーに対して照合されます。

「OAuth 2.0 トークンベース承認の使用」を参照してください。

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

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

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

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

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

「Service Registry を使用したスキーマの管理」を参照してください。

AMQ Streams で MirrorMaker 2.0 を使用できるようになりました。

MirrorMaker 2.0 は Kafka Connect フレームワークをベースとし、コネクターによってクラスター間のデータ転送が管理されます。

MirrorMaker 2.0 は以下を使用します。

MirrorMaker 2.0 では、クラスターでデータをレプリケートする全く新しい方法が導入されました。MirrorMaker 2.0 の使用を選択した場合、現在、レガシーサポートがないため、リソースを手作業で新しい形式に変換する必要があります。

「AMQ Streams の MirrorMaker 2.0 との使用」を参照してください。

OpenShift クラスターが制限されたネットワークで非接続クラスター として使用されている場合、AMQ Streams の 非接続インストール を実行できます。

非接続インストールでは、必要なイメージを取得し、それらをローカルでコンテナーレジストリーにプッシュします。Operator Lifecycle Manager (OLM) を使用している場合、OperatorHub によって使用されるデフォルトのソースを無効にし、ローカルミラーを作成してローカルソースから AMQ Streams をインストールすることになります。

「ネットワークが制限された環境での Operator Lifecycle Manager の使用」を参照してください。

ZooKeeper のスケールアップまたはダウンに関連する既知の問題があります。ZooKeeper のスケールアップとは、サーバーを ZooKeeper クラスターに追加することを言います。ZooKeeper のスケールダウンとは、ZooKeeper クラスターからサーバーを削除することを言います。

Kafka 2.4.0 には ZooKeeper 3.5.7 が必要です。

ZooKeeper 3.5.7 サーバーの設定手順は、ZooKeeper 3.4.x サーバーとは大きく異なります。新しい設定手順は 動的再設定 と呼ばれ、ZooKeeper CLI または Admin API を使用してサーバーを追加または削除する必要があります。これにより、スケールアップまたはダウンの処理中に ZooKeeper クラスターの安定性が維持されます。

ZooKeeper 3.5.7 クラスターをスケールアップまたはダウンするには、この既知の問題に記載されている手順を実行する必要があります。

各 ZooKeeper サーバーに対して、1 つずつ以下の手順を実行します。

ZooKeeper ノードを削除する際、番号が最も大きいサーバーから降順で削除されます。したがって、5 つのノードで構成されるクラスターを 3 つのノードにスケールダウンする場合、zookeeper-4 および zookeeper-3 が削除され、zookeeper-0、zookeeper-1、および zookeeper-2 が維持されます。

各 ZooKeeper サーバーに対して、1 つずつ以下の手順を実行します。