AMQ Streams 2.3 is supported on OpenShift Container Platform 4.8 to 4.12.

For more information about the supported platform versions, see the AMQ Streams Supported Configurations.

AMQ Streams now supports Apache Kafka version 3.3.1.

AMQ Streams uses Kafka 3.3.1. Only Kafka distributions built by Red Hat are supported.

You must upgrade the Cluster Operator to AMQ Streams version 2.3 before you can upgrade brokers and client applications to Kafka 3.3.1. For upgrade instructions, see Upgrading AMQ Streams.

Refer to the Kafka 3.3.0 and Kafka 3.3.1 Release Notes for additional information.

For more information on supported versions, see the AMQ Streams Component Details.

The v1beta2 API version for all custom resources was introduced with AMQ Streams 1.7. For AMQ Streams 1.8, v1alpha1 and v1beta1 API versions were removed from all AMQ Streams custom resources apart from KafkaTopic and KafkaUser.

Upgrade of the custom resources to v1beta2 prepares AMQ Streams for a move to Kubernetes CRD v1, which is required for Kubernetes v1.22.

If you are upgrading from an AMQ Streams version prior to version 1.7:

When using AMQ Streams with Cruise Control, you can now automate the process of approving the optimization proposals generated. You generate an optimization proposal using the KafkaRebalance custom resource. To enable auto-approval, you add strimzi.io/rebalance-auto-approval: "true" as an annotation to the KafkaRebalance custom resource before you generate the proposal.

With manual approval, you make another request when a generated proposal has a ProposalReady status. You approve the proposal by adding the strimzi.io/rebalance: approve annotation to the KafkaRebalance resource in the new request.

With automatic approval, the proposal is generated and approved to complete the rebalance in a single request.

See Generating optimization proposals.

The KafkaUser custom resource has been updated to make configuration of ACL lists easier to manage.

Previously, you configured operations for ACL rules separately for each resource using the operation property.

authorization:
  type: simple
  acls:
    - resource:
        type: topic
        name: my-topic
      operation: Read
    - resource:
        type: topic
        name: my-topic
      operation: Describe
    - resource:
        type: topic
        name: my-topic
      operation: Write
    - resource:
        type: topic
        name: my-topic
      operation: Create

A new operations property allows you to list multiple ACL operations as a single rule for the same resource.

authorization:
  type: simple
  acls:
    - resource:
        type: topic
        name: my-topic
      operations:
        - Read
        - Describe
        - Create
        - Write

The operation property for the old configuration format is deprecated, but still supported.

See ACLRule schema reference.

Listeners are used for client connection to Kafka brokers. They are configured in the Kafka resource using .spec.kafka.listeners properties.

A new cluster-ip type of internal listener exposes a Kafka cluster based on per-broker ClusterIP services.

#...
spec:
  kafka:
    #...
    listeners:
      - name: external-cluster-ip
        type: cluster-ip
        tls: false
        port: 9096
    #...

This is a useful option when you can’t route through the headless service or you wish to incorporate a custom access mechanism. For example, you might use this listener when building your own type of external listener for a specific Ingress controller or the Kubernetes Gateway API.

See GenericKafkaListener schema reference.

Use leader election to run multiple parallel replicas of the Cluster Operator. One replica is elected as the active leader and operates the deployed resources. The other replicas run in standby mode. Replicas are useful for high availability. Additional replicas safeguard against disruption caused by major failure. This is especially important since the introduction of StrimziPodSets, whereby AMQ Streams handles the creation and management of pods for Kafka clusters. The Cluster Operator is responsible for restarting the pods.

To enable leader election, the STRIMZI_LEADER_ELECTION_ENABLED environment variable for the Cluster Operator must be set to true (default). The environment variable is set, along with related environment variables, in the Deployment custom resource that is used to deploy the Cluster Operator. By default, AMQ Streams runs with a single Cluster Operator replica that is always the leader replica. To add more replicas, you update the spec.replicas value in the Deployment custom resource.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: strimzi-cluster-operator
  labels:
    app: strimzi
spec:
  replicas: 1
  # ...
    spec:
      # ...
      containers:
        - name: strimzi-cluster-operator
          image: registry.redhat.io/amq7/amq-streams-rhel8-operator:2.3.0
          # ...
          env:
            # ...
            - name: STRIMZI_LEADER_ELECTION_ENABLED
              value: "true"
            - name: STRIMZI_LEADER_ELECTION_LEASE_NAME
              value: "strimzi-cluster-operator"
            - name: STRIMZI_LEADER_ELECTION_LEASE_NAMESPACE
              valueFrom:
                fieldRef:
                  fieldPath: metadata.namespace
            - name: STRIMZI_LEADER_ELECTION_IDENTITY
              valueFrom:
                fieldRef:
  # ...

See Running multiple Cluster Operator replicas with leader election and Leader election environment variables.

AMQ Streams 2.3 is enabled to run on IBM Z and LinuxONE s390x architecture.

Support for IBM Z and LinuxONE applies to AMQ Streams running with Kafka on OpenShift Container Platform 4.10 and later.

AMQ Streams 2.3 is enabled to run on IBM Power ppc64le architecture.

Support for IBM Power applies to AMQ Streams running with Kafka on OpenShift Container Platform 4.9 and later.

The Red Hat build of Debezium is a distributed change data capture platform. It captures row-level changes in databases, creates change event records, and streams the records to Kafka topics. Debezium is built on Apache Kafka. You can deploy and integrate the Red Hat build of Debezium with AMQ Streams. Following a deployment of AMQ Streams, you deploy Debezium as a connector configuration through Kafka Connect. Debezium passes change event records to AMQ Streams on OpenShift. Applications can read these change event streams and access the change events in the order in which they occurred.

Debezium has multiple uses, including:

Debezium provides connectors (based on Kafka Connect) for the following common databases:

For more information on deploying Debezium with AMQ Streams, refer to the Red Hat build of Debezium documentation.

You can use the Red Hat build of Apicurio Registry as a centralized store of service schemas for data streaming. For Kafka, you can use the Red Hat build of Apicurio Registry to store Apache Avro or JSON schema.

Apicurio Registry provides a REST API and a Java REST client to register and query the schemas from client applications through server-side endpoints.

Using Apicurio Registry decouples the process of managing schemas from the configuration of client applications. You enable an application to use a schema from the registry by specifying its URL in the client code.

For example, the schemas to serialize and deserialize messages can be stored in the registry, which are then referenced from the applications that use them to ensure that the messages that they send and receive are compatible with those schemas.

Kafka client applications can push or pull their schemas from Apicurio Registry at runtime.

For more information on using the Red Hat build of Apicurio Registry with AMQ Streams, refer to the Red Hat build of Apicurio Registry documentation.

For an overview of the enhancements introduced with Kafka 3.3.0 and 3.3.1, refer to the Kafka 3.3.0 and Kafka 3.3.1 Release Notes.

The status of the KafkaConnector custom resource now shows NotReady if the connector or any associated tasks are reporting as FAILED. Previously, the custom resource would show READY even when the connector or a task had failed.

The ControlPlaneListener feature gate has moved to GA, which means it is now permanently enabled and cannot be disabled.

With ControlPlaneListener enabled, the connections between the Kafka controller and brokers use an internal control plane listener on port 9090.

See ControlPlaneListener feature gate.

The ServiceAccountPatching feature gate has moved to GA, which means it is now permanently enabled and cannot be disabled.

With ServiceAccountPatching enabled, the Cluster Operator always reconciles service accounts and updates them when needed. For example, when you change service account labels or annotations using the template property of a custom resource, the operator automatically updates them on the existing service account resources.

See ServiceAccountPatching feature gate.

The UseStrimziPodSets feature gate moves to a beta level of maturity, and is now enabled by default. This means StrimziPodSets are used by default instead of StatefulSets.

The feature gate controls a resource for managing pods called StrimziPodSet. AMQ Streams handles the creation and management of pods instead of OpenShift. Using StrimziPodSets instead of StatefulSets provides more control over the functionality.

See UseStrimziPodSets feature gate and Feature gate releases.

Rack awareness for Kafka Bridge pods is now supported. Use the KafkaBridge custom resource to configure rack awareness. You can configure a Kafka Bridge pod to be aware of the rack in which it runs. A rack can represent an availability zone, data center, or an actual rack in your data center.

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

See Rack schema reference.

Security context defines constraints on pods and containers. OpenShift uses built-in security context constraints (SCCs) to control permissions. SCCs are the settings and strategies that control the security features a pod has access to. You can also create and manage your own SCCs.

The optional STRIMZI_POD_SECURITY_PROVIDER_CLASS environment variable for the Cluster Operator provides security context configuration for pods and containers.

See Applying security context to AMQ Streams pods and containers.

After the Cluster Operator restarts a Kafka pod in an OpenShift cluster, it emits an OpenShift event into the pod’s namespace explaining why the pod restarted. For help in understanding cluster behavior, you can check the reason for a restart event from the command line. When checking restart events from the command line, you can also specify a reason or other field-selector options to filter the events returned.

The following example returns restart events that were triggered due to an error.

oc -n kafka get events --field-selector reportingController=strimzi.io/cluster-operator,reason=PodForceRestartOnError

See Finding information on Kafka restarts.

A new User Operator environment variable called STRIMZI_KAFKA_ADMIN_CLIENT_CONFIGURATION can pass additional configuration to the Kafka Admin client. The Kafka Admin client helps manage brokers and topics. You can now tune the Kafka Admin client without rebuilding the User Operator. For example, you can use it to pass SASL configuration or tune timeouts.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: strimzi-user-operator
  labels:
    app: strimzi
spec:
  # ...
  template:
    # ...
    spec:
      # ...
      containers:
        - name: strimzi-user-operator
          # ...
          env:
            - name: STRIMZI_KAFKA_ADMIN_CLIENT_CONFIGURATION
              value: |
                default.api.timeout.ms=120000
                request.timeout.ms=60000

See Deploying the standalone User Operator.

New Cruise Control configuration options allow you to specify overrides that set the network capacity and CPU limits for each Kafka broker. You can use these options when brokers are running on nodes with heterogeneous network or CPU resources.

Override capacity limits can be set for the following broker resources:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  # ...
  cruiseControl:
    # ...
    brokerCapacity:
      cpu: "1"
      inboundNetwork: 10000KiB/s
      outboundNetwork: 10000KiB/s
      overrides:
      - brokers: [0]
        cpu: "2.755"
        inboundNetwork: 20000KiB/s
        outboundNetwork: 20000KiB/s
      - brokers: [1, 2]
        cpu: 3000m
        inboundNetwork: 30000KiB/s
        outboundNetwork: 30000KiB/s

See Capacity overrides.

You can now configure Kafka clients to use the OAuth password grants mechanism for interaction with Kafka brokers.

security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.token.endpoint.uri="<token_endpoint_url>" \
  oauth.client.id="<client_id>" \ 1
  oauth.client.secret="<client_secret>" \ 2
  oauth.password.grant.username="<username>" \ 3
  oauth.password.grant.password="<password>" \ 4
  oauth.scope="<scope>" \
  oauth.audience="<audience>" ;
  # ...

See Configuring Kafka Java clients to use OAuth 2.0.

You can now collect metrics specific to oauth authentication and opa or keycloak authorization. You do this by setting the enableMetrics property to true in the listener configuration of the Kafka resource. For example, set enableMetrics to true in spec.kafka.listeners.authentication and spec.kafka.authorization. Similarly, you can enable metrics for oauth authentication in the KafkaBridge, KafkaConnect, KafkaMirrorMaker, and KafkaMirrorMaker2 custom resources.

See Introducing metrics.

This release introduces OpenTelemetry for distributed tracing as a technology preview. You can use OpenTelemetry with a specified tracing system. OpenTelemetry is replacing OpenTracing for distributed tracing. Support for OpenTracing is deprecated.

By Default, OpenTelemetry uses the OTLP (OpenTelemetry Protocol) exporter for tracing. AMQ Streams with OpenTelemetry is distributed for use with the Jaeger exporter, but you can specify other tracing systems supported by OpenTelemetry. AMQ Streams plans to migrate to using OpenTelemetry with the OTLP exporter by default, and is phasing out support for the Jaeger exporter.

See Introducing distributed tracing.

Use the Kafka Static Quota plugin to set throughput and storage limits on brokers in your Kafka cluster. You enable the plugin and set limits by configuring the Kafka resource. You can set a byte-rate threshold and storage quotas to put limits on the clients interacting with your brokers.

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    config:
      client.quota.callback.class: io.strimzi.kafka.quotas.StaticQuotaCallback
      client.quota.callback.static.produce: 1000000
      client.quota.callback.static.fetch: 1000000
      client.quota.callback.static.storage.soft: 400000000000
      client.quota.callback.static.storage.hard: 500000000000
      client.quota.callback.static.storage.check-interval: 5

See Setting limits on brokers using the Kafka Static Quota plugin.

As a Kafka cluster administrator, you can toggle a subset of features on and off using feature gates in the Cluster Operator deployment configuration.

Apache Kafka is in the process of phasing out the need for ZooKeeper. With the new UseKRaft feature gate enabled, you can try deploying a Kafka cluster in KRaft (Kafka Raft metadata) mode without ZooKeeper.

This feature gate is at an alpha level of maturity, and should be treated as a developer preview.

To enable the UseKRaft feature gate, specify +UseKRaft and +UseStrimziPodSets as values for the STRIMZI_FEATURE_GATES environment variable in the Cluster Operator configuration. The UseKRaft feature gate depends on the UseStrimziPodSets feature gate.

env:
  - name: STRIMZI_FEATURE_GATES
    value: +UseKRaft, +UseStrimziPodSets

Currently, the KRaft mode in AMQ Streams has the following major limitations:

See UseKRaft feature gate and Feature gate releases.

Kafka no longer includes the example file connectors FileStreamSourceConnector and FileStreamSinkConnector in its CLASSPATH and plugin.path by default. AMQ Streams has been updated so that you can still use these example connectors. The examples now have to be added to the plugin path like any connector.

Two example connector configuration files are provided:

See Deploying example KafkaConnector resources and Extending Kafka Connect with connector plugins.

Support for Java 8 was deprecated in Kafka 3.0.0 and AMQ Streams 2.0. Support for Java 8 will be removed in AMQ Streams 2.4.0. This applies to all AMQ Streams components, including clients.

AMQ Streams supports Java 11. Use Java 11 when developing new applications. Plan to migrate any applications that currently use Java 8 to Java 11.

If you want to continue using Java 8 for the time being, AMQ Streams 2.2 provides Long Term Support (LTS). For information on the LTS terms and dates, see the AMQ Streams LTS Support Policy.

Support for type: jaeger tracing is deprecated.

The Jaeger clients are now retired and the OpenTracing project archived. As such, we cannot guarantee their support for future Kafka versions. We are introducing a new tracing implementation based on the OpenTelemetry project.

The operation property for configuring operations for ACL rules is deprecated. A new, more-streamlined configuration format using the operations property is now available. For more information, see Section 1.5, “Support for multiple operations in ACL rule configuration”.

Kafka MirrorMaker replicates data between two or more active Kafka clusters, within or across data centers. Kafka MirrorMaker 1 is deprecated for Kafka 3.0.0 and will be removed in Kafka 4.0.0. MirrorMaker 2.0 will be the only version available. MirrorMaker 2.0 is based on the Kafka Connect framework, connectors managing the transfer of data between clusters.

As a consequence, the AMQ Streams KafkaMirrorMaker custom resource which is used to deploy Kafka MirrorMaker 1 has been deprecated. The KafkaMirrorMaker resource will be removed from AMQ Streams when Kafka 4.0.0 is adopted.

If you are using MirrorMaker 1 (referred to as just MirrorMaker in the AMQ Streams documentation), use the KafkaMirrorMaker2 custom resource with the IdentityReplicationPolicy. MirrorMaker 2.0 renames topics replicated to a target cluster. IdentityReplicationPolicy configuration overrides the automatic renaming. Use it to produce the same active/passive unidirectional replication as MirrorMaker 1.

See Kafka MirrorMaker 2.0 cluster configuration.

The Cruise Control TLS sidecar has been removed. As a result, the .spec.cruiseControl.tlsSidecar and .spec.cruiseControl.template.tlsSidecar properties are now deprecated. The properties are ignored and will be removed in the future.

Identity replication policy is used with MirrorMaker 2.0 to override the automatic renaming of remote topics. Instead of prepending the name with the name of the source cluster, the topic retains its original name. This optional setting is useful for active/passive backups and data migration.

The AMQ Streams Identity Replication Policy class (io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy) is now deprecated and will be removed in the future. You can update to use Kafka’s own Identity Replication Policy (class org.apache.kafka.connect.mirror.IdentityReplicationPolicy).

See Kafka MirrorMaker 2.0 cluster configuration.

The type property of ListenerStatus has been deprecated and will be removed in the future. ListenerStatus is used to specify the addresses of internal and external listeners. Instead of using the type, the addresses are now specified by name.

See ListenerStatus schema reference.

The disk and cpuUtilization capacity configuration properties have been deprecated, are ignored, and will be removed in the future. The properties were used in setting capacity limits in optimization proposals to determine if resource-based optimization goals are being broken. Disk and CPU capacity limits are now automatically generated by AMQ Streams.

See Configuring and deploying Cruise Control with Kafka.

If Cross-Origin Resource Sharing (CORS) is enabled for the Kafka Bridge, a 400 bad request error is returned when sending a HTTP request to produce messages.

To use CORS, you can deploy Red Hat 3scale for the Kafka Bridge.

The AMQ Streams Cluster Operator does not start on Internet Protocol version 6 (IPv6) clusters.

Cruise Control for AMQ Streams has a known issue that relates to the calculation of CPU utilization estimation. CPU utilization is calculated as a percentage of the defined capacity of a broker pod. The issue occurs when running Kafka brokers across nodes with varying CPU cores. For example, node1 might have 2 CPU cores and node2 might have 4 CPU cores. In this situation, Cruise Control can underestimate and overestimate CPU load of brokers The issue can prevent cluster rebalances when the pod is under heavy load.

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
  zookeeper:
    # ...
  entityOperator:
    topicOperator: {}
    userOperator: {}
  cruiseControl:
    brokerCapacity:
      inboundNetwork: 10000KB/s
      outboundNetwork: 10000KB/s
    config:
      hard.goals: >
        com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.MinTopicLeadersPerBrokerGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskCapacityGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundCapacityGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundCapacityGoal
      default.goals: >
        com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.MinTopicLeadersPerBrokerGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskCapacityGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundCapacityGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundCapacityGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaDistributionGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.PotentialNwOutGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskUsageDistributionGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundUsageDistributionGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundUsageDistributionGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.TopicReplicaDistributionGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.LeaderReplicaDistributionGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.LeaderBytesInDistributionGoal

For more information, see Insufficient CPU capacity.

The User Operator can timeout when creating multiple users at the same time. Reconciliation can take too long.

OAuth password grants are currently not being handled correctly by the Kafka Bridge. The OAuth authentication is not being configured properly.

This will be fixed for the next release.

Support for tracing using OpenTelemetry is built in to the following Kafka components:

When using the Jaeger exporter, trace data is retrieved through the Jaeger gPRC endpoint. By default, this endpoint does not have TLS enabled. However, it can still be configured to use TLS when deploying the Jaeger instance using the Jaeger operator. For example, when running the Red Hat OpenShift distributed tracing operator on OpenShift, which is a Jaeger operator, the operator automatically enables TLS. Jaeger instances with TLS on the gRPC endpoint are not supported on AMQ Streams.

There are two workarounds for this issue.

apiVersion: jaegertracing.io/v1
kind: Jaeger
metadata:
  name: my-jaeger
spec:
  allInOne:
    options:
      agent.grpc.tls.enabled: false
      collector.grpc.tls.enabled: false

In this example, the pipeline is from Jaeger and OTLP receivers to a Jaeger gRPC endpoint.

apiVersion: opentelemetry.io/v1alpha1
kind: OpenTelemetryCollector
metadata:
  name: cluster-collector
  namespace: <namespace>
spec:
  mode: deployment
  config: |
   receivers:
     otlp:
       protocols:
         grpc:
         http:
     jaeger:
       protocols:
         grpc:
   exporters:
     jaeger:
     endpoint: jaeger-all-in-one-inmemory-collector-headless.openshift-distributed-tracing.svc.cluster.local:14250
       tls:
         ca_file: "/var/run/secrets/kubernetes.io/serviceaccount/service-ca.crt"
   service:
     pipelines:
       traces:
         receivers: [otlp,jaeger]
         exporters: [jaeger]

To use the collector, you then need to specify the collector endpoint as the exporter endpoint in the tracing configuration.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
spec:
  #...
  template:
    connectContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_JAEGER_ENDPOINT
          value: "http:// jaeger-all-in-one-inmemory-collector-headless.openshift-distributed-tracing.svc.cluster.local:14250"
  tracing:
    type: opentelemetry
  #...