Configuring AMQ Streams on OpenShift

Red Hat AMQ Streams 2.3

Configure and manage a deployment of AMQ Streams 2.3 on OpenShift Container Platform

Abstract

Configure the operators and Kafka components deployed with AMQ Streams to build a large-scale messaging network.

Making open source more inclusive

Red Hat is committed to replacing problematic language in our code, documentation, and web properties. We are beginning with these four terms: master, slave, blacklist, and whitelist. Because of the enormity of this endeavor, these changes will be implemented gradually over several upcoming releases. For more details, see our CTO Chris Wright’s message.

Chapter 1. Configuration overview

AMQ Streams simplifies the process of running Apache Kafka in an OpenShift cluster.

This guide describes how to configure and manage an AMQ Streams deployment.

1.1. Configuring custom resources

Use custom resources to configure your AMQ Streams deployment.

You can use custom resources to configure and create instances of the following components:

Kafka clusters
Kafka Connect clusters
Kafka MirrorMaker
Kafka Bridge
Cruise Control

You can also use custom resource configuration to manage your instances or modify your deployment to introduce additional features. This might include configuration that supports the following:

Securing client access to Kafka brokers
Accessing Kafka brokers from outside the cluster
Creating topics
Creating users (clients)
Controlling feature gates
Changing logging frequency
Allocating resource limits and requests
Introducing features, such as AMQ Streams Drain Cleaner, Cruise Control, or distributed tracing.

The Custom resource API reference describes the properties you can use in your configuration.

1.2. Using ConfigMaps to add configuration

Use ConfigMap resources to add specific configuration to your AMQ Streams deployment. ConfigMaps use key-value pairs to store non-confidential data. Configuration data added to ConfigMaps is maintained in one place and can be reused amongst components.

ConfigMaps can only store configuration data related to the following:

Logging configuration
Metrics configuration
External configuration for Kafka Connect connectors

You can’t use ConfigMaps for other areas of configuration.

When you configure a component, you can add a reference to a ConfigMap using the configMapKeyRef property.

For example, you can use configMapKeyRef to reference a ConfigMap that provides configuration for logging. You might use a ConfigMap to pass a Log4j configuration file. You add the reference to the logging configuration.

Example ConfigMap for logging

spec:
  # ...
  logging:
    type: external
    valueFrom:
      configMapKeyRef:
        name: my-config-map
        key: my-config-map-key

To use a ConfigMap for metrics configuration, you add a reference to the metricsConfig configuration of the component in the same way.

ExternalConfiguration properties make data from a ConfigMap (or Secret) mounted to a pod available as environment variables or volumes. You can use external configuration data for the connectors used by Kafka Connect. The data might be related to an external data source, providing the values needed for the connector to communicate with that data source.

For example, you can use the configMapKeyRef property to pass configuration data from a ConfigMap as an environment variable.

Example ConfigMap providing environment variable values

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect
spec:
  # ...
  externalConfiguration:
    env:
      - name: MY_ENVIRONMENT_VARIABLE
        valueFrom:
          configMapKeyRef:
            name: my-config-map
            key: my-key

If you are using ConfigMaps that are managed externally, use configuration providers to load the data in the ConfigMaps. For more information on using configuration providers, see Chapter 3, Loading configuration values from external sources.

1.2.1. Naming custom ConfigMaps

AMQ Streams creates its own ConfigMaps and other resources when it is deployed to OpenShift. The ConfigMaps contain data necessary for running components. The ConfigMaps created by AMQ Streams must not be edited.

Make sure that any custom ConfigMaps you create do not have the same name as these default ConfigMaps. If they have the same name, they will be overwritten. For example, if your ConfigMap has the same name as the ConfigMap for the Kafka cluster, it will be overwritten when there is an update to the Kafka cluster.

Additional resources

Section 2.2.9, “List of Kafka cluster resources” (including ConfigMaps)
Section 2.10, “Logging configuration”
Section 12.1.8, “metricsConfig”
Section 12.2.74, “ExternalConfiguration schema reference”
Chapter 3, Loading configuration values from external sources

1.3. Configuring listeners to connect to Kafka brokers

Listeners are used for client connection to Kafka brokers. AMQ Streams provides a generic GenericKafkaListener schema with properties to configure listeners through the Kafka resource.

The GenericKafkaListener provides a flexible approach to listener configuration. You can specify properties to configure internal listeners for connecting within the OpenShift cluster or external listeners for connecting outside the OpenShift cluster.

Each listener is defined as an array in the Kafka resource. You can configure as many listeners as required, as long as their names and ports are unique. You can configure listeners for secure connection using authentication.

1.3.1. Configuring internal listeners

Internal listeners connect clients to Kafka brokers within the OpenShift cluster. An internal type listener configuration uses a headless service and the DNS names given to the broker pods.

You might need to join your OpenShift network to an outside network. In which case, you can configure an internal type listener (using the useServiceDnsDomain property) so that the OpenShift service DNS domain (typically .cluster.local) is not used.

You can also configure a cluster-ip type of listener that exposes a Kafka cluster based on per-broker ClusterIP services. 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 OpenShift Gateway API.

1.3.2. Configuring external listeners

Configure external listeners to handle access to a Kafka cluster from networks that require different authentication mechanisms.

You can configure external listeners for client access outside an OpenShift environment using a specified connection mechanism, such as a loadbalancer or route.

1.3.3. Providing listener certificates

You can provide your own server certificates, called Kafka listener certificates, for TLS listeners or external listeners which have TLS encryption enabled. For more information, see Kafka listener certificates.

Note

If you scale your Kafka cluster while using external listeners, it might trigger a rolling update of all Kafka brokers. This depends on the configuration.

Additional resources

Section 12.2.4, “GenericKafkaListener schema reference”
Chapter 5, Accessing Kafka outside of the OpenShift cluster
Section 6.3, “Securing access to Kafka brokers”

1.4. Document Conventions

User-replaced values

User-replaced values, also known as replaceables, are shown in italics with angle brackets (< >). Underscores ( _ ) are used for multi-word values. If the value refers to code or commands, monospace is also used.

For example, in the following code, you will want to replace <my_namespace> with the name of your namespace:

sed -i 's/namespace: .*/namespace: <my_namespace>/' install/cluster-operator/*RoleBinding*.yaml

1.5. Additional resources

Chapter 2. Configuring an AMQ Streams on OpenShift deployment

Configure your AMQ Streams deployment using custom resources. AMQ Streams provides example configuration files, which can serve as a starting point when building your own Kafka component configuration for deployment.

Note

Labels applied to a custom resource are also applied to the OpenShift resources making up its cluster. This provides a convenient mechanism for resources to be labeled as required.

Monitoring an AMQ Streams deployment

You can use Prometheus and Grafana to monitor your AMQ Streams deployment. For more information, see Introducing metrics to Kafka.

2.1. Using standard Kafka configuration properties

Use standard Kafka configuration properties to configure Kafka components.

The properties provide options to control and tune the configuration of the following Kafka components:

Brokers
Topics
Clients (producers and consumers)
Admin client
Kafka Connect
Kafka Streams

Broker and client parameters include options to configure authorization, authentication and encryption.

Note

For AMQ Streams on OpenShift, some configuration properties are managed entirely by AMQ Streams and cannot be changed.

For further information on Kafka configuration properties and how to use the properties to tune your deployment, see the following guides:

2.2. Kafka cluster configuration

Configure a Kafka deployment using the Kafka resource. A Kafka cluster is deployed with a ZooKeeper cluster, so configuration options are also available for ZooKeeper within the Kafka resource. The Entity Operator comprises the Topic Operator and User Operator. You can also configure entityOperator properties in the Kafka resource to include the Topic Operator and User Operator in the deployment.

Section 12.2.1, “Kafka schema reference” describes the full schema of the Kafka resource.

For more information about Apache Kafka, see the Apache Kafka documentation.

Listener configuration

You configure listeners for connecting clients to Kafka brokers. For more information on configuring listeners for connecting brokers, see Listener configuration.

Authorizing access to Kafka

You can configure your Kafka cluster to allow or decline actions executed by users. For more information, see Securing access to Kafka brokers.

Managing TLS certificates

When deploying Kafka, the Cluster Operator automatically sets up and renews TLS certificates to enable encryption and authentication within your cluster. If required, you can manually renew the cluster and clients CA certificates before their renewal period starts. You can also replace the keys used by the cluster and clients CA certificates. For more information, see Renewing CA certificates manually and Replacing private keys.

2.2.1. Configuring Kafka

Use the properties of the Kafka resource to configure your Kafka deployment.

As well as configuring Kafka, you can add configuration for ZooKeeper and the AMQ Streams Operators. Common configuration properties, such as logging and healthchecks, are configured independently for each component.

This procedure shows only some of the possible configuration options, but those that are particularly important include:

Resource requests (CPU / Memory)
JVM options for maximum and minimum memory allocation
Listeners (and authentication of clients)
Authentication
Storage
Rack awareness
Metrics
Cruise Control for cluster rebalancing

Kafka versions

The inter.broker.protocol.version property for the Kafka config must be the version supported by the specified Kafka version (spec.kafka.version). The property represents the version of Kafka protocol used in a Kafka cluster.

From Kafka 3.0.0, when the inter.broker.protocol.version is set to 3.0 or higher, the log.message.format.version option is ignored and doesn’t need to be set.

An update to the inter.broker.protocol.version is required when upgrading your Kafka version. For more information, see Upgrading Kafka.

Prerequisites

An OpenShift cluster
A running Cluster Operator

See the Deploying and Upgrading AMQ Streams on OpenShift guide for instructions on deploying a:

Procedure

Edit the spec properties for the Kafka resource.
The properties you can configure are shown in this example configuration:
```
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    replicas: 3 1
    version: 3.3.1 2
    logging: 3
      type: inline
      loggers:
        kafka.root.logger.level: "INFO"
    resources: 4
      requests:
        memory: 64Gi
        cpu: "8"
      limits:
        memory: 64Gi
        cpu: "12"
    readinessProbe: 5
      initialDelaySeconds: 15
      timeoutSeconds: 5
    livenessProbe:
      initialDelaySeconds: 15
      timeoutSeconds: 5
    jvmOptions: 6
      -Xms: 8192m
      -Xmx: 8192m
    image: my-org/my-image:latest 7
    listeners: 8
      - name: plain 9
        port: 9092 10
        type: internal 11
        tls: false 12
        configuration:
          useServiceDnsDomain: true 13
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication: 14
          type: tls
      - name: external 15
        port: 9094
        type: route
        tls: true
        configuration:
          brokerCertChainAndKey: 16
            secretName: my-secret
            certificate: my-certificate.crt
            key: my-key.key
    authorization: 17
      type: simple
    config: 18
      auto.create.topics.enable: "false"
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
      inter.broker.protocol.version: "3.3"
      ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 19
      ssl.enabled.protocols: "TLSv1.2"
      ssl.protocol: "TLSv1.2"
    storage: 20
      type: persistent-claim 21
      size: 10000Gi 22
    rack: 23
      topologyKey: topology.kubernetes.io/zone
    metricsConfig: 24
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef: 25
          name: my-config-map
          key: my-key
    # ...
  zookeeper: 26
    replicas: 3 27
    logging: 28
      type: inline
      loggers:
        zookeeper.root.logger: "INFO"
    resources:
      requests:
        memory: 8Gi
        cpu: "2"
      limits:
        memory: 8Gi
        cpu: "2"
    jvmOptions:
      -Xms: 4096m
      -Xmx: 4096m
    storage:
      type: persistent-claim
      size: 1000Gi
    metricsConfig:
      # ...
  entityOperator: 29
    tlsSidecar: 30
      resources:
        requests:
          cpu: 200m
          memory: 64Mi
        limits:
          cpu: 500m
          memory: 128Mi
    topicOperator:
      watchedNamespace: my-topic-namespace
      reconciliationIntervalSeconds: 60
      logging: 31
        type: inline
        loggers:
          rootLogger.level: "INFO"
      resources:
        requests:
          memory: 512Mi
          cpu: "1"
        limits:
          memory: 512Mi
          cpu: "1"
    userOperator:
      watchedNamespace: my-topic-namespace
      reconciliationIntervalSeconds: 60
      logging: 32
        type: inline
        loggers:
          rootLogger.level: INFO
      resources:
        requests:
          memory: 512Mi
          cpu: "1"
        limits:
          memory: 512Mi
          cpu: "1"
  kafkaExporter: 33
    # ...
  cruiseControl: 34
    # ...
```
1
The number of replica nodes. If your cluster already has topics defined, you can scale clusters.
2
Kafka version, which can be changed to a supported version by following the upgrade procedure.
3
Kafka loggers and log levels added directly (inline) or indirectly (external) through a ConfigMap. A custom ConfigMap must be placed under the log4j.properties key. For the Kafka kafka.root.logger.level logger, you can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.
4
Requests for reservation of supported resources, currently cpu and memory, and limits to specify the maximum resources that can be consumed.
5
Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).
6
JVM configuration options to optimize performance for the Virtual Machine (VM) running Kafka.
7
ADVANCED OPTION: Container image configuration, which is recommended only in special situations.
8
Listeners configure how clients connect to the Kafka cluster via bootstrap addresses. Listeners are configured as internal or external listeners for connection from inside or outside the OpenShift cluster.
9
Name to identify the listener. Must be unique within the Kafka cluster.
10
Port number used by the listener inside Kafka. The port number has to be unique within a given Kafka cluster. Allowed port numbers are 9092 and higher with the exception of ports 9404 and 9999, which are already used for Prometheus and JMX. Depending on the listener type, the port number might not be the same as the port number that connects Kafka clients.
11
Listener type specified as internal or cluster-ip (to expose Kafka using per-broker ClusterIP services), or for external listeners, as route, loadbalancer, nodeport or ingress.
12
Enables TLS encryption for each listener. Default is false. TLS encryption is not required for route listeners.
13
Defines whether the fully-qualified DNS names including the cluster service suffix (usually .cluster.local) are assigned.
14
Listener authentication mechanism specified as mTLS, SCRAM-SHA-512, or token-based OAuth 2.0.
15
External listener configuration specifies how the Kafka cluster is exposed outside OpenShift, such as through a route, loadbalancer or nodeport.
16
Optional configuration for a Kafka listener certificate managed by an external CA (certificate authority). The brokerCertChainAndKey specifies a Secret that contains a server certificate and a private key. You can configure Kafka listener certificates on any listener with enabled TLS encryption.
17
Authorization enables simple, OAUTH 2.0, or OPA authorization on the Kafka broker. Simple authorization uses the AclAuthorizer Kafka plugin.
18
Broker configuration. Standard Apache Kafka configuration may be provided, restricted to those properties not managed directly by AMQ Streams.
19
SSL properties for listeners with TLS encryption enabled to enable a specific cipher suite or TLS version.
20
Storage is configured as ephemeral, persistent-claim or jbod.
21
Storage size for persistent volumes may be increased and additional volumes may be added to JBOD storage.
22
Persistent storage has additional configuration options, such as a storage id and class for dynamic volume provisioning.
23
Rack awareness configuration to spread replicas across different racks, data centers, or availability zones. The topologyKey must match a node label containing the rack ID. The example used in this configuration specifies a zone using the standard topology.kubernetes.io/zone label.
24
Prometheus metrics enabled. In this example, metrics are configured for the Prometheus JMX Exporter (the default metrics exporter).
25
Prometheus rules for exporting metrics to a Grafana dashboard through the Prometheus JMX Exporter, which are enabled by referencing a ConfigMap containing configuration for the Prometheus JMX exporter. You can enable metrics without further configuration using a reference to a ConfigMap containing an empty file under metricsConfig.valueFrom.configMapKeyRef.key.
26
ZooKeeper-specific configuration, which contains properties similar to the Kafka configuration.
27
The number of ZooKeeper nodes. ZooKeeper clusters or ensembles usually run with an odd number of nodes, typically three, five, or seven. The majority of nodes must be available in order to maintain an effective quorum. If the ZooKeeper cluster loses its quorum, it will stop responding to clients and the Kafka brokers will stop working. Having a stable and highly available ZooKeeper cluster is crucial for AMQ Streams.
28
Specified ZooKeeper loggers and log levels.
29
Entity Operator configuration, which specifies the configuration for the Topic Operator and User Operator.
30
Entity Operator TLS sidecar configuration. Entity Operator uses the TLS sidecar for secure communication with ZooKeeper.
31
Specified Topic Operator loggers and log levels. This example uses inline logging.
32
Specified User Operator loggers and log levels.
33
Kafka Exporter configuration. Kafka Exporter is an optional component for extracting metrics data from Kafka brokers, in particular consumer lag data. For Kafka Exporter to be able to work properly, consumer groups need to be in use.
34
Optional configuration for Cruise Control, which is used to rebalance the Kafka cluster.
Create or update the resource:
```
oc apply -f <kafka_configuration_file>
```

2.2.2. Configuring the Entity Operator

The Entity Operator is responsible for managing Kafka-related entities in a running Kafka cluster.

The Entity Operator comprises the:

Topic Operator to manage Kafka topics
User Operator to manage Kafka users

Through Kafka resource configuration, the Cluster Operator can deploy the Entity Operator, including one or both operators, when deploying a Kafka cluster.

The operators are automatically configured to manage the topics and users of the Kafka cluster. The Topic Operator and User Operator can only watch a single namespace. For more information, see Section 7.1, “Watching namespaces with AMQ Streams operators”.

Note

When deployed, the Entity Operator pod contains the operators according to the deployment configuration.

2.2.2.1. Entity Operator configuration properties

Use the entityOperator property in Kafka.spec to configure the Entity Operator.

The entityOperator property supports several sub-properties:

tlsSidecar
topicOperator
userOperator
template

The tlsSidecar property contains the configuration of the TLS sidecar container, which is used to communicate with ZooKeeper.

The template property contains the configuration of the Entity Operator pod, such as labels, annotations, affinity, and tolerations. For more information on configuring templates, see Section 2.8, “Customizing OpenShift resources”.

The topicOperator property contains the configuration of the Topic Operator. When this option is missing, the Entity Operator is deployed without the Topic Operator.

The userOperator property contains the configuration of the User Operator. When this option is missing, the Entity Operator is deployed without the User Operator.

For more information on the properties used to configure the Entity Operator, see the EntityUserOperatorSpec schema reference.

Example of basic configuration enabling both operators

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

If an empty object ({}) is used for the topicOperator and userOperator, all properties use their default values.

When both topicOperator and userOperator properties are missing, the Entity Operator is not deployed.

2.2.2.2. Topic Operator configuration properties

Topic Operator deployment can be configured using additional options inside the topicOperator object. The following properties are supported:

watchedNamespace: The OpenShift namespace in which the Topic Operator watches for KafkaTopic resources. Default is the namespace where the Kafka cluster is deployed.
reconciliationIntervalSeconds: The interval between periodic reconciliations in seconds. Default 120.
zookeeperSessionTimeoutSeconds: The ZooKeeper session timeout in seconds. Default 18.
topicMetadataMaxAttempts: The number of attempts at getting topic metadata from Kafka. The time between each attempt is defined as an exponential back-off. Consider increasing this value when topic creation might take more time due to the number of partitions or replicas. Default 6.
image: The image property can be used to configure the container image which will be used. For more details about configuring custom container images, see Section 12.1.6, “image”.
resources: The resources property configures the amount of resources allocated to the Topic Operator. For more details about resource request and limit configuration, see Section 12.1.5, “resources”.
logging: The logging property configures the logging of the Topic Operator. For more details, see Section 12.2.45.1, “logging”.

Example Topic Operator configuration

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
  zookeeper:
    # ...
  entityOperator:
    # ...
    topicOperator:
      watchedNamespace: my-topic-namespace
      reconciliationIntervalSeconds: 60
    # ...

2.2.2.3. User Operator configuration properties

User Operator deployment can be configured using additional options inside the userOperator object. The following properties are supported:

watchedNamespace: The OpenShift namespace in which the User Operator watches for KafkaUser resources. Default is the namespace where the Kafka cluster is deployed.
reconciliationIntervalSeconds: The interval between periodic reconciliations in seconds. Default 120.
image: The image property can be used to configure the container image which will be used. For more details about configuring custom container images, see Section 12.1.6, “image”.
resources: The resources property configures the amount of resources allocated to the User Operator. For more details about resource request and limit configuration, see Section 12.1.5, “resources”.
logging: The logging property configures the logging of the User Operator. For more details, see Section 12.2.45.1, “logging”.
secretPrefix: The secretPrefix property adds a prefix to the name of all Secrets created from the KafkaUser resource. For example, secretPrefix: kafka- would prefix all Secret names with kafka-. So a KafkaUser named my-user would create a Secret named kafka-my-user.

Example User Operator configuration

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
  zookeeper:
    # ...
  entityOperator:
    # ...
    userOperator:
      watchedNamespace: my-user-namespace
      reconciliationIntervalSeconds: 60
    # ...

2.2.3. Configuring Kafka and ZooKeeper storage

As stateful applications, Kafka and ZooKeeper store data on disk. AMQ Streams supports three storage types for this data:

Ephemeral (Recommended for development only)
Persistent
JBOD (Kafka only not ZooKeeper)

When configuring a Kafka resource, you can specify the type of storage used by the Kafka broker and its corresponding ZooKeeper node. You configure the storage type using the storage property in the following resources:

Kafka.spec.kafka
Kafka.spec.zookeeper

The storage type is configured in the type field.

Refer to the schema reference for more information on storage configuration properties:

EphemeralStorage schema reference
PersistentClaimStorage schema reference
JbodStorage schema reference

Warning

The storage type cannot be changed after a Kafka cluster is deployed.

2.2.3.1. Data storage considerations

For AMQ Streams to work well, an efficient data storage infrastructure is essential. Block storage is required. File storage, such as NFS, does not work with Kafka.

Choose one of the following options for your block storage:

A cloud-based block storage solution, such as Amazon Elastic Block Store (EBS)
Persistent storage using local persistent volumes
Storage Area Network (SAN) volumes accessed by a protocol such as Fibre Channel or iSCSI

Note

AMQ Streams does not require OpenShift raw block volumes.

2.2.3.1.1. File systems

Kafka uses a file system for storing messages. AMQ Streams is compatible with the XFS and ext4 file systems, which are commonly used with Kafka. Consider the underlying architecture and requirements of your deployment when choosing and setting up your file system.

For more information, refer to Filesystem Selection in the Kafka documentation.

2.2.3.1.2. Disk usage

Use separate disks for Apache Kafka and ZooKeeper.

Solid-state drives (SSDs), though not essential, can improve the performance of Kafka in large clusters where data is sent to and received from multiple topics asynchronously. SSDs are particularly effective with ZooKeeper, which requires fast, low latency data access.

Note

You do not need to provision replicated storage because Kafka and ZooKeeper both have built-in data replication.

2.2.3.2. Ephemeral storage

Ephemeral data storage is transient. All pods on a node share a local ephemeral storage space. Data is retained for as long as the pod that uses it is running. The data is lost when a pod is deleted. Although a pod can recover data in a highly available environment.

Because of its transient nature, ephemeral storage is only recommended for development and testing.

Ephemeral storage uses emptyDir volumes to store data. An emptyDir volume is created when a pod is assigned to a node. You can set the total amount of storage for the emptyDir using the sizeLimit property .

Important

Ephemeral storage is not suitable for single-node ZooKeeper clusters or Kafka topics with a replication factor of 1.

To use ephemeral storage, you set the storage type configuration in the Kafka or ZooKeeper resource to ephemeral.

Example ephemeral storage configuration

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    storage:
      type: ephemeral
    # ...
  zookeeper:
    # ...
    storage:
      type: ephemeral
    # ...

2.2.3.2.1. Mount path of Kafka log directories

The ephemeral volume is used by Kafka brokers as log directories mounted into the following path:

/var/lib/kafka/data/kafka-logIDX

Where IDX is the Kafka broker pod index. For example /var/lib/kafka/data/kafka-log0.

2.2.3.3. Persistent storage

Persistent data storage retains data in the event of system disruption. For pods that use persistent data storage, data is persisted across pod failures and restarts.

A dynamic provisioning framework enables clusters to be created with persistent storage. Pod configuration uses Persistent Volume Claims (PVCs) to make storage requests on persistent volumes (PVs). PVs are storage resources that represent a storage volume. PVs are independent of the pods that use them. The PVC requests the amount of storage required when a pod is being created. The underlying storage infrastructure of the PV does not need to be understood. If a PV matches the storage criteria, the PVC is bound to the PV.

Because of its permanent nature, persistent storage is recommended for production.

PVCs can request different types of persistent storage by specifying a StorageClass. Storage classes define storage profiles and dynamically provision PVs. If a storage class is not specified, the default storage class is used. Persistent storage options might include SAN storage types or local persistent volumes.

To use persistent storage, you set the storage type configuration in the Kafka or ZooKeeper resource to persistent-claim.

In the production environment, the following configuration is recommended:

For Kafka, configure type: jbod with one or more type: persistent-claim volumes
For ZooKeeper, configure type: persistent-claim

Persistent storage also has the following configuration options:

id (optional): A storage identification number. This option is mandatory for storage volumes defined in a JBOD storage declaration. Default is 0.
size (required): The size of the persistent volume claim, for example, "1000Gi".
class (optional): The OpenShift StorageClass to use for dynamic volume provisioning. Storage class configuration includes parameters that describe the profile of a volume in detail.
selector (optional): Configuration to specify a specific PV. Provides key:value pairs representing the labels of the volume selected.
deleteClaim (optional): Boolean value to specify whether the PVC is deleted when the cluster is uninstalled. Default is false.

Warning

Increasing the size of persistent volumes in an existing AMQ Streams cluster is only supported in OpenShift versions that support persistent volume resizing. The persistent volume to be resized must use a storage class that supports volume expansion. For other versions of OpenShift and storage classes that do not support volume expansion, you must decide the necessary storage size before deploying the cluster. Decreasing the size of existing persistent volumes is not possible.

Example persistent storage configuration for Kafka and ZooKeeper

# ...
spec:
  kafka:
    # ...
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
      - id: 1
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
      - id: 2
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
    # ...
  zookeeper:
    storage:
      type: persistent-claim
      size: 1000Gi
# ...

If you do not specify a storage class, the default is used. The following example specifies a storage class.

Example persistent storage configuration with specific storage class

# ...
storage:
  type: persistent-claim
  size: 1Gi
  class: my-storage-class
# ...

Use a selector to specify a labeled persistent volume that provides certain features, such as an SSD.

Example persistent storage configuration with selector

# ...
storage:
  type: persistent-claim
  size: 1Gi
  selector:
    hdd-type: ssd
  deleteClaim: true
# ...

2.2.3.3.1. Storage class overrides

Instead of using the default storage class, you can specify a different storage class for one or more Kafka brokers or ZooKeeper nodes. This is useful, for example, when storage classes are restricted to different availability zones or data centers. You can use the overrides field for this purpose.

In this example, the default storage class is named my-storage-class:

Example AMQ Streams cluster using storage class overrides

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  labels:
    app: my-cluster
  name: my-cluster
  namespace: myproject
spec:
  # ...
  kafka:
    replicas: 3
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
        class: my-storage-class
        overrides:
        - broker: 0
          class: my-storage-class-zone-1a
        - broker: 1
          class: my-storage-class-zone-1b
        - broker: 2
          class: my-storage-class-zone-1c
      # ...
  # ...
  zookeeper:
    replicas: 3
    storage:
      deleteClaim: true
      size: 100Gi
      type: persistent-claim
      class: my-storage-class
      overrides:
        - broker: 0
          class: my-storage-class-zone-1a
        - broker: 1
          class: my-storage-class-zone-1b
        - broker: 2
          class: my-storage-class-zone-1c
  # ...

As a result of the configured overrides property, the volumes use the following storage classes:

The persistent volumes of ZooKeeper node 0 use my-storage-class-zone-1a.
The persistent volumes of ZooKeeper node 1 use my-storage-class-zone-1b.
The persistent volumes of ZooKeeepr node 2 use my-storage-class-zone-1c.
The persistent volumes of Kafka broker 0 use my-storage-class-zone-1a.
The persistent volumes of Kafka broker 1 use my-storage-class-zone-1b.
The persistent volumes of Kafka broker 2 use my-storage-class-zone-1c.

The overrides property is currently used only to override storage class configurations. Overrides for other storage configuration properties is not currently supported. Other storage configuration properties are currently not supported.

2.2.3.3.2. PVC resources for persistent storage

When persistent storage is used, it creates PVCs with the following names:

data-cluster-name-kafka-idx: PVC for the volume used for storing data for the Kafka broker pod idx.
data-cluster-name-zookeeper-idx: PVC for the volume used for storing data for the ZooKeeper node pod idx.

2.2.3.3.3. Mount path of Kafka log directories

The persistent volume is used by the Kafka brokers as log directories mounted into the following path:

/var/lib/kafka/data/kafka-logIDX

Where IDX is the Kafka broker pod index. For example /var/lib/kafka/data/kafka-log0.

2.2.3.4. Resizing persistent volumes

You can provision increased storage capacity by increasing the size of the persistent volumes used by an existing AMQ Streams cluster. Resizing persistent volumes is supported in clusters that use either a single persistent volume or multiple persistent volumes in a JBOD storage configuration.

Note

You can increase but not decrease the size of persistent volumes. Decreasing the size of persistent volumes is not currently supported in OpenShift.

Prerequisites

An OpenShift cluster with support for volume resizing.
The Cluster Operator is running.
A Kafka cluster using persistent volumes created using a storage class that supports volume expansion.

Procedure

Edit the Kafka resource for your cluster.
Change the size property to increase the size of the persistent volume allocated to a Kafka cluster, a ZooKeeper cluster, or both.
- For Kafka clusters, update the size property under spec.kafka.storage.
- For ZooKeeper clusters, update the size property under spec.zookeeper.storage.
Kafka configuration to increase the volume size to 2000Gi
```
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    storage:
      type: persistent-claim
      size: 2000Gi
      class: my-storage-class
    # ...
  zookeeper:
    # ...
```
Create or update the resource:
```
oc apply -f <kafka_configuration_file>
```
OpenShift increases the capacity of the selected persistent volumes in response to a request from the Cluster Operator. When the resizing is complete, the Cluster Operator restarts all pods that use the resized persistent volumes. This happens automatically.

Verify that the storage capacity has increased for the relevant pods on the cluster:

oc get pv

Kafka broker pods with increased storage

NAME               CAPACITY   CLAIM
pvc-0ca459ce-...   2000Gi     my-project/data-my-cluster-kafka-2
pvc-6e1810be-...   2000Gi     my-project/data-my-cluster-kafka-0
pvc-82dc78c9-...   2000Gi     my-project/data-my-cluster-kafka-1

The output shows the names of each PVC associated with a broker pod.

Additional resources

For more information about resizing persistent volumes in OpenShift, see Resizing Persistent Volumes using Kubernetes.

2.2.3.5. JBOD storage

You can configure AMQ Streams to use JBOD, a data storage configuration of multiple disks or volumes. JBOD is one approach to providing increased data storage for Kafka brokers. It can also improve performance.

Note

JBOD storage is supported for Kafka only not ZooKeeper.

A JBOD configuration is described by one or more volumes, each of which can be either ephemeral or persistent. The rules and constraints for JBOD volume declarations are the same as those for ephemeral and persistent storage. For example, you cannot decrease the size of a persistent storage volume after it has been provisioned, or you cannot change the value of sizeLimit when the type is ephemeral.

To use JBOD storage, you set the storage type configuration in the Kafka resource to jbod. The volumes property allows you to describe the disks that make up your JBOD storage array or configuration.

Example JBOD storage configuration

# ...
storage:
  type: jbod
  volumes:
  - id: 0
    type: persistent-claim
    size: 100Gi
    deleteClaim: false
  - id: 1
    type: persistent-claim
    size: 100Gi
    deleteClaim: false
# ...

The IDs cannot be changed once the JBOD volumes are created. You can add or remove volumes from the JBOD configuration.

2.2.3.5.1. PVC resource for JBOD storage

When persistent storage is used to declare JBOD volumes, it creates a PVC with the following name:

data-id-cluster-name-kafka-idx: PVC for the volume used for storing data for the Kafka broker pod idx. The id is the ID of the volume used for storing data for Kafka broker pod.

2.2.3.5.2. Mount path of Kafka log directories

The JBOD volumes are used by Kafka brokers as log directories mounted into the following path:

/var/lib/kafka/data-id/kafka-logidx

Where id is the ID of the volume used for storing data for Kafka broker pod idx. For example /var/lib/kafka/data-0/kafka-log0.

2.2.3.6. Adding volumes to JBOD storage

This procedure describes how to add volumes to a Kafka cluster configured to use JBOD storage. It cannot be applied to Kafka clusters configured to use any other storage type.

Note

When adding a new volume under an id which was already used in the past and removed, you have to make sure that the previously used PersistentVolumeClaims have been deleted.

Prerequisites

An OpenShift cluster
A running Cluster Operator
A Kafka cluster with JBOD storage

Procedure

Edit the spec.kafka.storage.volumes property in the Kafka resource. Add the new volumes to the volumes array. For example, add the new volume with id 2:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
      - id: 1
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
      - id: 2
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
    # ...
  zookeeper:
    # ...

Create or update the resource:
```
oc apply -f <kafka_configuration_file>
```
Create new topics or reassign existing partitions to the new disks.

Additional resources

For more information about reassigning topics, see Section 2.2.4.2, “Partition reassignment tool”.

2.2.3.7. Removing volumes from JBOD storage

This procedure describes how to remove volumes from Kafka cluster configured to use JBOD storage. It cannot be applied to Kafka clusters configured to use any other storage type. The JBOD storage always has to contain at least one volume.

Important

To avoid data loss, you have to move all partitions before removing the volumes.

Prerequisites

An OpenShift cluster
A running Cluster Operator
A Kafka cluster with JBOD storage with two or more volumes

Procedure

Reassign all partitions from the disks which are you going to remove. Any data in partitions still assigned to the disks which are going to be removed might be lost.

Edit the spec.kafka.storage.volumes property in the Kafka resource. Remove one or more volumes from the volumes array. For example, remove the volumes with ids 1 and 2:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
    # ...
  zookeeper:
    # ...

Create or update the resource:
```
oc apply -f <kafka_configuration_file>
```

Additional resources

For more information about reassigning topics, see Section 2.2.4.2, “Partition reassignment tool”.

2.2.4. Scaling clusters

Scale Kafka clusters by adding or removing brokers. If a cluster already has topics defined, you also have to reassign partitions.

Use the kafka-reassign-partitions.sh tool to reassign partitions. The tool uses a reassignment JSON file that specifies the topics to reassign.

You can generate a reassignment JSON file or create a file manually if you want to move specific partitions.

2.2.4.1. Broker scaling configuration

You configure the Kafka.spec.kafka.replicas configuration to add or reduce the number of brokers.

Broker addition

The primary way of increasing throughput for a topic is to increase the number of partitions for that topic. That works because the extra partitions allow the load of the topic to be shared between the different brokers in the cluster. However, in situations where every broker is constrained by a particular resource (typically I/O) using more partitions will not result in increased throughput. Instead, you need to add brokers to the cluster.

When you add an extra broker to the cluster, Kafka does not assign any partitions to it automatically. You must decide which partitions to reassign from the existing brokers to the new broker.

Once the partitions have been redistributed between all the brokers, the resource utilization of each broker is reduced.

Broker removal

If you are using StatefulSets to manage broker pods, you cannot remove any pod from the cluster. You can only remove one or more of the highest numbered pods from the cluster. For example, in a cluster of 12 brokers the pods are named cluster-name-kafka-0 up to cluster-name-kafka-11. If you decide to scale down by one broker, the cluster-name-kafka-11 will be removed.

Before you remove a broker from a cluster, ensure that it is not assigned to any partitions. You should also decide which of the remaining brokers will be responsible for each of the partitions on the broker being decommissioned. Once the broker has no assigned partitions, you can scale the cluster down safely.

2.2.4.2. Partition reassignment tool

The Topic Operator does not currently support reassigning replicas to different brokers, so it is necessary to connect directly to broker pods to reassign replicas to brokers.

Within a broker pod, the kafka-reassign-partitions.sh tool allows you to reassign partitions to different brokers.

It has three different modes:

--generate: Takes a set of topics and brokers and generates a reassignment JSON file which will result in the partitions of those topics being assigned to those brokers. Because this operates on whole topics, it cannot be used when you only want to reassign some partitions of some topics.
--execute: Takes a reassignment JSON file and applies it to the partitions and brokers in the cluster. Brokers that gain partitions as a result become followers of the partition leader. For a given partition, once the new broker has caught up and joined the ISR (in-sync replicas) the old broker will stop being a follower and will delete its replica.
--verify: Using the same reassignment JSON file as the --execute step, --verify checks whether all the partitions in the file have been moved to their intended brokers. If the reassignment is complete, --verify also removes any traffic throttles (--throttle) that are in effect. Unless removed, throttles will continue to affect the cluster even after the reassignment has finished.

It is only possible to have one reassignment running in a cluster at any given time, and it is not possible to cancel a running reassignment. If you need to cancel a reassignment, wait for it to complete and then perform another reassignment to revert the effects of the first reassignment. The kafka-reassign-partitions.sh will print the reassignment JSON for this reversion as part of its output. Very large reassignments should be broken down into a number of smaller reassignments in case there is a need to stop in-progress reassignment.

2.2.4.2.1. Partition reassignment JSON file

The reassignment JSON file has a specific structure:

{
  "version": 1,
  "partitions": [
    <PartitionObjects>
  ]
}

Where <PartitionObjects> is a comma-separated list of objects like:

{
  "topic": <TopicName>,
  "partition": <Partition>,
  "replicas": [ <AssignedBrokerIds> ]
}

Note

Although Kafka also supports a "log_dirs" property this should not be used in AMQ Streams.

The following is an example reassignment JSON file that assigns partition 4 of topic topic-a to brokers 2, 4 and 7, and partition 2 of topic topic-b to brokers 1, 5 and 7:

Example partition reassignment file

{
  "version": 1,
  "partitions": [
    {
      "topic": "topic-a",
      "partition": 4,
      "replicas": [2,4,7]
    },
    {
      "topic": "topic-b",
      "partition": 2,
      "replicas": [1,5,7]
    }
  ]
}

Partitions not included in the JSON are not changed.

2.2.4.2.2. Partition reassignment between JBOD volumes

When using JBOD storage in your Kafka cluster, you can choose to reassign the partitions between specific volumes and their log directories (each volume has a single log directory). To reassign a partition to a specific volume, add the log_dirs option to <PartitionObjects> in the reassignment JSON file.

{
  "topic": <TopicName>,
  "partition": <Partition>,
  "replicas": [ <AssignedBrokerIds> ],
  "log_dirs": [ <AssignedLogDirs> ]
}

The log_dirs object should contain the same number of log directories as the number of replicas specified in the replicas object. The value should be either an absolute path to the log directory, or the any keyword.

Example partition reassignment file specifying log directories

{
      "topic": "topic-a",
      "partition": 4,
      "replicas": [2,4,7].
      "log_dirs": [ "/var/lib/kafka/data-0/kafka-log2", "/var/lib/kafka/data-0/kafka-log4", "/var/lib/kafka/data-0/kafka-log7" ]
}

Partition reassignment throttles

Partition reassignment can be a slow process because it involves transferring large amounts of data between brokers. To avoid a detrimental impact on clients, you can throttle the reassignment process. Use the --throttle parameter with the kafka-reassign-partitions.sh tool to throttle a reassignment. You specify a maximum threshold in bytes per second for the movement of partitions between brokers. For example, --throttle 5000000 sets a maximum threshold for moving partitions of 50 MBps.

Throttling might cause the reassignment to take longer to complete.

If the throttle is too low, the newly assigned brokers will not be able to keep up with records being published and the reassignment will never complete.
If the throttle is too high, clients will be impacted.

For example, for producers, this could manifest as higher than normal latency waiting for acknowledgment. For consumers, this could manifest as a drop in throughput caused by higher latency between polls.

2.2.4.3. Generating reassignment JSON files

This procedure describes how to generate a reassignment JSON file. Use the reassignment file with the kafka-reassign-partitions.sh tool to reassign partitions after scaling a Kafka cluster.

You run the tool from an interactive pod container connected to the Kafka cluster.

The steps describe a secure reassignment process that uses mTLS. You’ll need a Kafka cluster that uses TLS encryption and mTLS authentication.

You’ll need the following to establish a connection:

The cluster CA certificate and password generated by the Cluster Operator when the Kafka cluster is created
The user CA certificate and password generated by the User Operator when a user is created for client access to the Kafka cluster

In this procedure, the CA certificates and corresponding passwords are extracted from the cluster and user secrets that contain them in PKCS #12 (.p12 and .password) format. The passwords allow access to the .p12 stores that contain the certificates. You use the .p12 stores to specify a truststore and keystore to authenticate connection to the Kafka cluster.

Prerequisites

You have a running Cluster Operator.
You have a running Kafka cluster based on a Kafka resource configured with internal TLS encryption and mTLS authentication.
Kafka configuration with TLS encryption and mTLS authentication
```
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    listeners:
      # ...
      - name: tls
        port: 9093
        type: internal
        tls: true 1
        authentication:
          type: tls 2
    # ...
```
1
Enables TLS encryption for the internal listener.
2
Listener authentication mechanism specified as mutual tls.

The running Kafka cluster contains a set of topics and partitions to reassign.

Example topic configuration for my-topic

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: my-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 10
  replicas: 3
  config:
    retention.ms: 7200000
    segment.bytes: 1073741824
    # ...

You have a KafkaUser configured with ACL rules that specify permission to produce and consume topics from the Kafka brokers.

Example Kafka user configuration with ACL rules to allow operations on my-topic and my-cluster

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication: 1
    type: tls
  authorization:
    type: simple 2
    acls:
      # access to the topic
      - resource:
          type: topic
          name: my-topic
        operations:
          - Create
          - Describe
          - Read
          - AlterConfigs
        host: "*"
      # access to the cluster
      - resource:
          type: cluster
        operations:
          - Alter
          - AlterConfigs
        host: "*"
      # ...
  # ...

1: User authentication mechanism defined as mutual tls.
2: Simple authorization and accompanying list of ACL rules.

Procedure

Extract the cluster CA certificate and password from the <cluster_name>-cluster-ca-cert secret of the Kafka cluster.
```
oc get secret <cluster_name>-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12
```
```
oc get secret <cluster_name>-cluster-ca-cert -o jsonpath='{.data.ca\.password}' | base64 -d > ca.password
```
Replace <cluster_name> with the name of the Kafka cluster. When you deploy Kafka using the Kafka resource, a secret with the cluster CA certificate is created with the Kafka cluster name (<cluster_name>-cluster-ca-cert). For example, my-cluster-cluster-ca-cert.
Run a new interactive pod container using the AMQ Streams Kafka image to connect to a running Kafka broker.
```
oc run --restart=Never --image=registry.redhat.io/amq7/amq-streams-kafka-33-rhel8:2.3.0 <interactive_pod_name> -- /bin/sh -c "sleep 3600"
```
Replace <interactive_pod_name> with the name of the pod.
Copy the cluster CA certificate to the interactive pod container.
```
oc cp ca.p12 <interactive_pod_name>:/tmp
```
Extract the user CA certificate and password from the secret of the Kafka user that has permission to access the Kafka brokers.
```
oc get secret <kafka_user> -o jsonpath='{.data.user\.p12}' | base64 -d > user.p12
```
```
oc get secret <kafka_user> -o jsonpath='{.data.user\.password}' | base64 -d > user.password
```
Replace <kafka_user> with the name of the Kafka user. When you create a Kafka user using the KafkaUser resource, a secret with the user CA certificate is created with the Kafka user name. For example, my-user.
Copy the user CA certificate to the interactive pod container.
```
oc cp user.p12 <interactive_pod_name>:/tmp
```
The CA certificates allow the interactive pod container to connect to the Kafka broker using TLS.
Create a config.properties file to specify the truststore and keystore used to authenticate connection to the Kafka cluster.
Use the certificates and passwords you extracted in the previous steps.
```
bootstrap.servers=<kafka_cluster_name>-kafka-bootstrap:9093 1
security.protocol=SSL 2
ssl.truststore.location=/tmp/ca.p12 3
ssl.truststore.password=<truststore_password> 4
ssl.keystore.location=/tmp/user.p12 5
ssl.keystore.password=<keystore_password> 6
```
1
The bootstrap server address to connect to the Kafka cluster. Use your own Kafka cluster name to replace <kafka_cluster_name>.
2
The security protocol option when using TLS for encryption.
3
The truststore location contains the public key certificate (ca.p12) for the Kafka cluster.
4
The password (ca.password) for accessing the truststore.
5
The keystore location contains the public key certificate (user.p12) for the Kafka user.
6
The password (user.password) for accessing the keystore.

Copy the config.properties file to the interactive pod container.

oc cp config.properties <interactive_pod_name>:/tmp/config.properties

Prepare a JSON file named topics.json that specifies the topics to move.
Specify topic names as a comma-separated list.
Example JSON file to reassign all the partitions of topic-a and topic-b
```
{
  "version": 1,
  "topics": [
    { "topic": "topic-a"},
    { "topic": "topic-b"}
  ]
}
```

Copy the topics.json file to the interactive pod container.

oc cp topics.json <interactive_pod_name>:/tmp/topics.json

Start a shell process in the interactive pod container.
```
oc exec -n <namespace> -ti <interactive_pod_name> /bin/bash
```
Replace <namespace> with the OpenShift namespace where the pod is running.

Use the kafka-reassign-partitions.sh command to generate the reassignment JSON.

Example command to move all the partitions of topic-a and topic-b to brokers 0, 1 and 2

bin/kafka-reassign-partitions.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 \
  --command-config /tmp/config.properties \
  --topics-to-move-json-file /tmp/topics.json \
  --broker-list 0,1,2 \
  --generate

Additional resources

Section 2.2.1, “Configuring Kafka”
Section 7.3.3, “Configuring Kafka topics”
Section 7.4.1, “Configuring Kafka users”

2.2.4.4. Scaling up a Kafka cluster

Use a reassignment file to increase the number of brokers in a Kafka cluster.

The reassignment file should describe how partitions are reassigned to brokers in the enlarged Kafka cluster.

This procedure describes a secure scaling process that uses TLS. You’ll need a Kafka cluster that uses TLS encryption and mTLS authentication.

Prerequisites

You have a running Kafka cluster based on a Kafka resource configured with internal TLS encryption and mTLS authentication.
You have generated a reassignment JSON file named reassignment.json.
You are running an interactive pod container that is connected to the running Kafka broker.
You are connected as a KafkaUser configured with ACL rules that specify permission to manage the Kafka cluster and its topics.

See Generating reassignment JSON files.

Procedure

Add as many new brokers as you need by increasing the Kafka.spec.kafka.replicas configuration option.
Verify that the new broker pods have started.
If you haven’t done so, run an interactive pod container to generate a reassignment JSON file named reassignment.json.
Copy the reassignment.json file to the interactive pod container.
```
oc cp reassignment.json <interactive_pod_name>:/tmp/reassignment.json
```
Replace <interactive_pod_name> with the name of the pod.
Start a shell process in the interactive pod container.
```
oc exec -n <namespace> -ti <interactive_pod_name> /bin/bash
```
Replace <namespace> with the OpenShift namespace where the pod is running.
Run the partition reassignment using the kafka-reassign-partitions.sh script from the interactive pod container.
```
bin/kafka-reassign-partitions.sh --bootstrap-server
 <cluster_name>-kafka-bootstrap:9093 \
 --command-config /tmp/config.properties \
 --reassignment-json-file /tmp/reassignment.json \
 --execute
```
Replace <cluster_name> with the name of your Kafka cluster. For example, my-cluster-kafka-bootstrap:9093
If you are going to throttle replication, you can also pass the --throttle option with an inter-broker throttled rate in bytes per second. For example:
```
bin/kafka-reassign-partitions.sh --bootstrap-server
  <cluster_name>-kafka-bootstrap:9093 \
  --command-config /tmp/config.properties \
  --reassignment-json-file /tmp/reassignment.json \
  --throttle 5000000 \
  --execute
```
This command will print out two reassignment JSON objects. The first records the current assignment for the partitions being moved. You should save this to a local file (not a file in the pod) in case you need to revert the reassignment later on. The second JSON object is the target reassignment you have passed in your reassignment JSON file.
If you need to change the throttle during reassignment, you can use the same command with a different throttled rate. For example:
```
bin/kafka-reassign-partitions.sh --bootstrap-server
  <cluster_name>-kafka-bootstrap:9093 \
  --command-config /tmp/config.properties \
  --reassignment-json-file /tmp/reassignment.json \
  --throttle 10000000 \
  --execute
```
Verify that the reassignment has completed using the kafka-reassign-partitions.sh command line tool from any of the broker pods. This is the same command as the previous step, but with the --verify option instead of the --execute option.
```
bin/kafka-reassign-partitions.sh --bootstrap-server
  <cluster_name>-kafka-bootstrap:9093 \
  --command-config /tmp/config.properties \
  --reassignment-json-file /tmp/reassignment.json \
  --verify
```
The reassignment has finished when the --verify command reports that each of the partitions being moved has completed successfully. This final --verify will also have the effect of removing any reassignment throttles.
You can now delete the revert file if you saved the JSON for reverting the assignment to their original brokers.

2.2.4.5. Scaling down a Kafka cluster

Use a reassignment file to decrease the number of brokers in a Kafka cluster.

The reassignment file must describe how partitions are reassigned to the remaining brokers in the Kafka cluster. Brokers in the highest numbered pods are removed first.

This procedure describes a secure scaling process that uses TLS. You’ll need a Kafka cluster that uses TLS encryption and mTLS authentication.

Prerequisites

You have a running Kafka cluster based on a Kafka resource configured with internal TLS encryption and mTLS authentication.
You have generated a reassignment JSON file named reassignment.json.
You are running an interactive pod container that is connected to the running Kafka broker.
You are connected as a KafkaUser configured with ACL rules that specify permission to manage the Kafka cluster and its topics.

See Generating reassignment JSON files.

Procedure

If you haven’t done so, run an interactive pod container to generate a reassignment JSON file named reassignment.json.
Copy the reassignment.json file to the interactive pod container.
```
oc cp reassignment.json <interactive_pod_name>:/tmp/reassignment.json
```
Replace <interactive_pod_name> with the name of the pod.
Start a shell process in the interactive pod container.
```
oc exec -n <namespace> -ti <interactive_pod_name> /bin/bash
```
Replace <namespace> with the OpenShift namespace where the pod is running.
Run the partition reassignment using the kafka-reassign-partitions.sh script from the interactive pod container.
```
bin/kafka-reassign-partitions.sh --bootstrap-server
 <cluster_name>-kafka-bootstrap:9093 \
 --command-config /tmp/config.properties \
 --reassignment-json-file /tmp/reassignment.json \
 --execute
```
Replace <cluster_name> with the name of your Kafka cluster. For example, my-cluster-kafka-bootstrap:9093
If you are going to throttle replication, you can also pass the --throttle option with an inter-broker throttled rate in bytes per second. For example:
```
bin/kafka-reassign-partitions.sh --bootstrap-server
  <cluster_name>-kafka-bootstrap:9093 \
  --command-config /tmp/config.properties \
  --reassignment-json-file /tmp/reassignment.json \
  --throttle 5000000 \
  --execute
```
This command will print out two reassignment JSON objects. The first records the current assignment for the partitions being moved. You should save this to a local file (not a file in the pod) in case you need to revert the reassignment later on. The second JSON object is the target reassignment you have passed in your reassignment JSON file.
If you need to change the throttle during reassignment, you can use the same command with a different throttled rate. For example:
```
bin/kafka-reassign-partitions.sh --bootstrap-server
  <cluster_name>-kafka-bootstrap:9093 \
  --command-config /tmp/config.properties \
  --reassignment-json-file /tmp/reassignment.json \
  --throttle 10000000 \
  --execute
```
Verify that the reassignment has completed using the kafka-reassign-partitions.sh command line tool from any of the broker pods. This is the same command as the previous step, but with the --verify option instead of the --execute option.
```
bin/kafka-reassign-partitions.sh --bootstrap-server
  <cluster_name>-kafka-bootstrap:9093 \
  --command-config /tmp/config.properties \
  --reassignment-json-file /tmp/reassignment.json \
  --verify
```
The reassignment has finished when the --verify command reports that each of the partitions being moved has completed successfully. This final --verify will also have the effect of removing any reassignment throttles.
You can now delete the revert file if you saved the JSON for reverting the assignment to their original brokers.
When all the partition reassignments have finished, the brokers being removed should not have responsibility for any of the partitions in the cluster. You can verify this by checking that the broker’s data log directory does not contain any live partition logs. If the log directory on the broker contains a directory that does not match the extended regular expression \.[a-z0-9]-delete$, the broker still has live partitions and should not be stopped.
You can check this by executing the command:
```
oc exec my-cluster-kafka-0 -c kafka -it -- \
  /bin/bash -c \
  "ls -l /var/lib/kafka/kafka-log_<n>_ | grep -E '^d' | grep -vE '[a-zA-Z0-9.-]+\.[a-z0-9]+-delete$'"
```
where n is the number of the pods being deleted.
If the above command prints any output then the broker still has live partitions. In this case, either the reassignment has not finished or the reassignment JSON file was incorrect.
When you have confirmed that the broker has no live partitions, you can edit the Kafka.spec.kafka.replicas property of your Kafka resource to reduce the number of brokers.

2.2.5. Maintenance time windows for rolling updates

Maintenance time windows allow you to schedule certain rolling updates of your Kafka and ZooKeeper clusters to start at a convenient time.

2.2.5.1. Maintenance time windows overview

In most cases, the Cluster Operator only updates your Kafka or ZooKeeper clusters in response to changes to the corresponding Kafka resource. This enables you to plan when to apply changes to a Kafka resource to minimize the impact on Kafka client applications.

However, some updates to your Kafka and ZooKeeper clusters can happen without any corresponding change to the Kafka resource. For example, the Cluster Operator will need to perform a rolling restart if a CA (certificate authority) certificate that it manages is close to expiry.

While a rolling restart of the pods should not affect availability of the service (assuming correct broker and topic configurations), it could affect performance of the Kafka client applications. Maintenance time windows allow you to schedule such spontaneous rolling updates of your Kafka and ZooKeeper clusters to start at a convenient time. If maintenance time windows are not configured for a cluster then it is possible that such spontaneous rolling updates will happen at an inconvenient time, such as during a predictable period of high load.

2.2.5.2. Maintenance time window definition

You configure maintenance time windows by entering an array of strings in the Kafka.spec.maintenanceTimeWindows property. Each string is a cron expression interpreted as being in UTC (Coordinated Universal Time, which for practical purposes is the same as Greenwich Mean Time).

The following example configures a single maintenance time window that starts at midnight and ends at 01:59am (UTC), on Sundays, Mondays, Tuesdays, Wednesdays, and Thursdays:

# ...
maintenanceTimeWindows:
  - "* * 0-1 ? * SUN,MON,TUE,WED,THU *"
# ...

In practice, maintenance windows should be set in conjunction with the Kafka.spec.clusterCa.renewalDays and Kafka.spec.clientsCa.renewalDays properties of the Kafka resource, to ensure that the necessary CA certificate renewal can be completed in the configured maintenance time windows.

Note

AMQ Streams does not schedule maintenance operations exactly according to the given windows. Instead, for each reconciliation, it checks whether a maintenance window is currently "open". This means that the start of maintenance operations within a given time window can be delayed by up to the Cluster Operator reconciliation interval. Maintenance time windows must therefore be at least this long.

Additional resources

For more information about the Cluster Operator configuration, see Section 7.2.3, “Configuring the Cluster Operator with environment variables”.

2.2.5.3. Configuring a maintenance time window

You can configure a maintenance time window for rolling updates triggered by supported processes.

Prerequisites

An OpenShift cluster.
The Cluster Operator is running.

Procedure

Add or edit the maintenanceTimeWindows property in the Kafka resource. For example to allow maintenance between 0800 and 1059 and between 1400 and 1559 you would set the maintenanceTimeWindows as shown below:
```
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
  zookeeper:
    # ...
  maintenanceTimeWindows:
    - "* * 8-10 * * ?"
    - "* * 14-15 * * ?"
```
Create or update the resource:
```
oc apply -f <kafka_configuration_file>
```

Additional resources

Performing rolling updates:

Section 11.4.2, “Performing a rolling update using a pod management annotation”
Section 11.4.3, “Performing a rolling update using a Pod annotation”

2.2.6. Connecting to ZooKeeper from a terminal

Most Kafka CLI tools can connect directly to Kafka, so under normal circumstances you should not need to connect to ZooKeeper. ZooKeeper services are secured with encryption and authentication and are not intended to be used by external applications that are not part of AMQ Streams.

However, if you want to use Kafka CLI tools that require a connection to ZooKeeper, you can use a terminal inside a ZooKeeper container and connect to localhost:12181 as the ZooKeeper address.

Prerequisites

An OpenShift cluster is available.
A Kafka cluster is running.
The Cluster Operator is running.

Procedure

Open the terminal using the OpenShift console or run the exec command from your CLI.
For example:
```
oc exec -ti my-cluster-zookeeper-0 -- bin/kafka-topics.sh --list --zookeeper localhost:12181
```
Be sure to use localhost:12181.
You can now run Kafka commands to ZooKeeper.

2.2.7. Deleting Kafka nodes manually

This procedure describes how to delete an existing Kafka node by using an OpenShift annotation. Deleting a Kafka node consists of deleting both the Pod on which the Kafka broker is running and the related PersistentVolumeClaim (if the cluster was deployed with persistent storage). After deletion, the Pod and its related PersistentVolumeClaim are recreated automatically.

Warning

Deleting a PersistentVolumeClaim can cause permanent data loss. The following procedure should only be performed if you have encountered storage issues.

Prerequisites

See the Deploying and Upgrading AMQ Streams on OpenShift guide for instructions on running a:

Procedure

Find the name of the Pod that you want to delete.
Kafka broker pods are named <cluster-name>-kafka-<index>, where <index> starts at zero and ends at the total number of replicas minus one. For example, my-cluster-kafka-0.

Annotate the Pod resource in OpenShift.

Use oc annotate:

oc annotate pod cluster-name-kafka-index strimzi.io/delete-pod-and-pvc=true

Wait for the next reconciliation, when the annotated pod with the underlying persistent volume claim will be deleted and then recreated.

2.2.8. Deleting ZooKeeper nodes manually

This procedure describes how to delete an existing ZooKeeper node by using an OpenShift annotation. Deleting a ZooKeeper node consists of deleting both the Pod on which ZooKeeper is running and the related PersistentVolumeClaim (if the cluster was deployed with persistent storage). After deletion, the Pod and its related PersistentVolumeClaim are recreated automatically.

Warning

Deleting a PersistentVolumeClaim can cause permanent data loss. The following procedure should only be performed if you have encountered storage issues.

Prerequisites

See the Deploying and Upgrading AMQ Streams on OpenShift guide for instructions on running a:

Procedure

Find the name of the Pod that you want to delete.
ZooKeeper pods are named <cluster-name>-zookeeper-<index>, where <index> starts at zero and ends at the total number of replicas minus one. For example, my-cluster-zookeeper-0.

Annotate the Pod resource in OpenShift.

Use oc annotate:

oc annotate pod cluster-name-zookeeper-index strimzi.io/delete-pod-and-pvc=true

Wait for the next reconciliation, when the annotated pod with the underlying persistent volume claim will be deleted and then recreated.

2.2.9. List of Kafka cluster resources

The following resources are created by the Cluster Operator in the OpenShift cluster:

Shared resources

cluster-name-cluster-ca: Secret with the Cluster CA private key used to encrypt the cluster communication.
cluster-name-cluster-ca-cert: Secret with the Cluster CA public key. This key can be used to verify the identity of the Kafka brokers.
cluster-name-clients-ca: Secret with the Clients CA private key used to sign user certificates
cluster-name-clients-ca-cert: Secret with the Clients CA public key. This key can be used to verify the identity of the Kafka users.
cluster-name-cluster-operator-certs: Secret with Cluster operators keys for communication with Kafka and ZooKeeper.

ZooKeeper nodes

cluster-name-zookeeper

Name given to the following ZooKeeper resources:

StatefulSet or StrimziPodSet (if the UseStrimziPodSets feature gate is enabled) for managing the ZooKeeper node pods.
Service account used by the ZooKeeper nodes.
PodDisruptionBudget configured for the ZooKeeper nodes.

cluster-name-zookeeper-idx

Pods created by the ZooKeeper StatefulSet or StrimziPodSet.

cluster-name-zookeeper-nodes

Headless Service needed to have DNS resolve the ZooKeeper pods IP addresses directly.

cluster-name-zookeeper-client

Service used by Kafka brokers to connect to ZooKeeper nodes as clients.

cluster-name-zookeeper-config

ConfigMap that contains the ZooKeeper ancillary configuration, and is mounted as a volume by the ZooKeeper node pods.

cluster-name-zookeeper-nodes

Secret with ZooKeeper node keys.

cluster-name-network-policy-zookeeper

Network policy managing access to the ZooKeeper services.

data-cluster-name-zookeeper-idx

Persistent Volume Claim for the volume used for storing data for the ZooKeeper node pod idx. This resource will be created only if persistent storage is selected for provisioning persistent volumes to store data.

Kafka brokers

cluster-name-kafka

Name given to the following Kafka resources:

StatefulSet or StrimziPodSet (if the UseStrimziPodSets feature gate is enabled) for managing the Kafka broker pods.
Service account used by the Kafka pods.
PodDisruptionBudget configured for the Kafka brokers.

cluster-name-kafka-idx

Name given to the following Kafka resources:

Pods created by the Kafka StatefulSet or StrimziPodSet.
ConfigMap with Kafka broker configuration (if the UseStrimziPodSets feature gate is enabled).

cluster-name-kafka-brokers

Service needed to have DNS resolve the Kafka broker pods IP addresses directly.

cluster-name-kafka-bootstrap

Service can be used as bootstrap servers for Kafka clients connecting from within the OpenShift cluster.

cluster-name-kafka-external-bootstrap

Bootstrap service for clients connecting from outside the OpenShift cluster. This resource is created only when an external listener is enabled. The old service name will be used for backwards compatibility when the listener name is external and port is 9094.

cluster-name-kafka-pod-id

Service used to route traffic from outside the OpenShift cluster to individual pods. This resource is created only when an external listener is enabled. The old service name will be used for backwards compatibility when the listener name is external and port is 9094.

cluster-name-kafka-external-bootstrap

Bootstrap route for clients connecting from outside the OpenShift cluster. This resource is created only when an external listener is enabled and set to type route. The old route name will be used for backwards compatibility when the listener name is external and port is 9094.

cluster-name-kafka-pod-id

Route for traffic from outside the OpenShift cluster to individual pods. This resource is created only when an external listener is enabled and set to type route. The old route name will be used for backwards compatibility when the listener name is external and port is 9094.

cluster-name-kafka-listener-name-bootstrap

Bootstrap service for clients connecting from outside the OpenShift cluster. This resource is created only when an external listener is enabled. The new service name will be used for all other external listeners.

cluster-name-kafka-listener-name-pod-id

Service used to route traffic from outside the OpenShift cluster to individual pods. This resource is created only when an external listener is enabled. The new service name will be used for all other external listeners.

cluster-name-kafka-listener-name-bootstrap

Bootstrap route for clients connecting from outside the OpenShift cluster. This resource is created only when an external listener is enabled and set to type route. The new route name will be used for all other external listeners.

cluster-name-kafka-listener-name-pod-id

Route for traffic from outside the OpenShift cluster to individual pods. This resource is created only when an external listener is enabled and set to type route. The new route name will be used for all other external listeners.

cluster-name-kafka-config

ConfigMap which contains the Kafka ancillary configuration and is mounted as a volume by the Kafka broker pods.

cluster-name-kafka-brokers

Secret with Kafka broker keys.

cluster-name-network-policy-kafka

Network policy managing access to the Kafka services.

strimzi-namespace-name-cluster-name-kafka-init

Cluster role binding used by the Kafka brokers.

cluster-name-jmx

Secret with JMX username and password used to secure the Kafka broker port. This resource is created only when JMX is enabled in Kafka.

data-cluster-name-kafka-idx

Persistent Volume Claim for the volume used for storing data for the Kafka broker pod idx. This resource is created only if persistent storage is selected for provisioning persistent volumes to store data.

data-id-cluster-name-kafka-idx

Persistent Volume Claim for the volume id used for storing data for the Kafka broker pod idx. This resource is created only if persistent storage is selected for JBOD volumes when provisioning persistent volumes to store data.

Entity Operator

These resources are only created if the Entity Operator is deployed using the Cluster Operator.

cluster-name-entity-operator

Name given to the following Entity Operator resources:

Deployment with Topic and User Operators.
Service account used by the Entity Operator.

cluster-name-entity-operator-random-string

Pod created by the Entity Operator deployment.

cluster-name-entity-topic-operator-config

ConfigMap with ancillary configuration for Topic Operators.

cluster-name-entity-user-operator-config

ConfigMap with ancillary configuration for User Operators.

cluster-name-entity-topic-operator-certs

Secret with Topic Operator keys for communication with Kafka and ZooKeeper.

cluster-name-entity-user-operator-certs

Secret with User Operator keys for communication with Kafka and ZooKeeper.

strimzi-cluster-name-entity-topic-operator

Role binding used by the Entity Topic Operator.

strimzi-cluster-name-entity-user-operator

Role binding used by the Entity User Operator.

Kafka Exporter

These resources are only created if the Kafka Exporter is deployed using the Cluster Operator.

cluster-name-kafka-exporter

Name given to the following Kafka Exporter resources:

Deployment with Kafka Exporter.
Service used to collect consumer lag metrics.
Service account used by the Kafka Exporter.

cluster-name-kafka-exporter-random-string

Pod created by the Kafka Exporter deployment.

Cruise Control

These resources are only created if Cruise Control was deployed using the Cluster Operator.

cluster-name-cruise-control

Name given to the following Cruise Control resources:

Deployment with Cruise Control.
Service used to communicate with Cruise Control.
Service account used by the Cruise Control.

cluster-name-cruise-control-random-string

Pod created by the Cruise Control deployment.

cluster-name-cruise-control-config

ConfigMap that contains the Cruise Control ancillary configuration, and is mounted as a volume by the Cruise Control pods.

cluster-name-cruise-control-certs

Secret with Cruise Control keys for communication with Kafka and ZooKeeper.

cluster-name-network-policy-cruise-control

Network policy managing access to the Cruise Control service.

2.3. Kafka Connect cluster configuration

Configure a Kafka Connect deployment using the KafkaConnect resource. Kafka Connect is an integration toolkit for streaming data between Kafka brokers and other systems using connector plugins. Kafka Connect provides a framework for integrating Kafka with an external data source or target, such as a database, for import or export of data using connectors. Connectors are plugins that provide the connection configuration needed.

Section 12.2.60, “KafkaConnect schema reference” describes the full schema of the KafkaConnect resource.

For more information on deploying connector plugins, see Extending Kafka Connect with connector plugins.

2.3.1. Configuring Kafka Connect

Use Kafka Connect to set up external data connections to your Kafka cluster. Use the properties of the KafkaConnect resource to configure your Kafka Connect deployment.

KafkaConnector configuration

KafkaConnector resources allow you to create and manage connector instances for Kafka Connect in an OpenShift-native way.

In your Kafka Connect configuration, you enable KafkaConnectors for a Kafka Connect cluster by adding the strimzi.io/use-connector-resources annotation. You can also add a build configuration so that AMQ Streams automatically builds a container image with the connector plugins you require for your data connections. External configuration for Kafka Connect connectors is specified through the externalConfiguration property.

To manage connectors, you can use the Kafka Connect REST API, or use KafkaConnector custom resources. KafkaConnector resources must be deployed to the same namespace as the Kafka Connect cluster they link to. For more information on using these methods to create, reconfigure, or delete connectors, see Creating and managing connectors.

Connector configuration is passed to Kafka Connect as part of an HTTP request and stored within Kafka itself. ConfigMaps and Secrets are standard OpenShift resources used for storing configurations and confidential data. You can use ConfigMaps and Secrets to configure certain elements of a connector. You can then reference the configuration values in HTTP REST commands, which keeps the configuration separate and more secure, if needed. This method applies especially to confidential data, such as usernames, passwords, or certificates.

Handling high volumes of messages

You can tune the configuration to handle high volumes of messages. For more information, see Section 2.7, “Handling high volumes of messages”.

Prerequisites

An OpenShift cluster
A running Cluster Operator

See the Deploying and Upgrading AMQ Streams on OpenShift guide for instructions on running a:

Procedure

Edit the spec properties of the KafkaConnect resource.

The properties you can configure are shown in this example configuration:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect 1
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true" 2
spec:
  replicas: 3 3
  authentication: 4
    type: tls
    certificateAndKey:
      certificate: source.crt
      key: source.key
      secretName: my-user-source
  bootstrapServers: my-cluster-kafka-bootstrap:9092 5
  tls: 6
    trustedCertificates:
      - secretName: my-cluster-cluster-cert
        certificate: ca.crt
      - secretName: my-cluster-cluster-cert
        certificate: ca2.crt
  config: 7
    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
  build: 8
    output: 9
      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins: 10
      - 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
  externalConfiguration: 11
    env:
      - name: AWS_ACCESS_KEY_ID
        valueFrom:
          secretKeyRef:
            name: aws-creds
            key: awsAccessKey
      - name: AWS_SECRET_ACCESS_KEY
        valueFrom:
          secretKeyRef:
            name: aws-creds
            key: awsSecretAccessKey
  resources: 12
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  logging: 13
    type: inline
    loggers:
      log4j.rootLogger: "INFO"
  readinessProbe: 14
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  metricsConfig: 15
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        name: my-config-map
        key: my-key
  jvmOptions: 16
    "-Xmx": "1g"
    "-Xms": "1g"
  image: my-org/my-image:latest 17
  rack:
    topologyKey: topology.kubernetes.io/zone 18
  template: 19
    pod:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: application
                    operator: In
                    values:
                      - postgresql
                      - mongodb
              topologyKey: "kubernetes.io/hostname"
    connectContainer: 20
      env:
        - name: JAEGER_SERVICE_NAME
          value: my-jaeger-service
        - name: JAEGER_AGENT_HOST
          value: jaeger-agent-name
        - name: JAEGER_AGENT_PORT
          value: "6831"

1: Use KafkaConnect.
2: Enables KafkaConnectors for the Kafka Connect cluster.
3: The number of replica nodes for the workers that run tasks.
4: Authentication for the Kafka Connect cluster, specified as mTLS, token-based OAuth, SASL-based SCRAM-SHA-256/SCRAM-SHA-512, or PLAIN. By default, Kafka Connect connects to Kafka brokers using a plain text connection.
5: Bootstrap server for connection to the Kafka Connect cluster.
6: TLS encryption with key names under which TLS certificates are stored in X.509 format for the cluster. If certificates are stored in the same secret, it can be listed multiple times.
7: Kafka Connect configuration of workers (not connectors). Standard Apache Kafka configuration may be provided, restricted to those properties not managed directly by AMQ Streams.
8: Build configuration properties for building a container image with connector plugins automatically.
9: (Required) Configuration of the container registry where new images are pushed.
10: (Required) List of connector plugins and their artifacts to add to the new container image. Each plugin must be configured with at least one artifact.
11: External configuration for Kafka connectors using environment variables, as shown here, or volumes. You can also use configuration provider plugins to load configuration values from external sources.
12: Requests for reservation of supported resources, currently cpu and memory, and limits to specify the maximum resources that can be consumed.
13: Specified Kafka Connect loggers and log levels added directly (inline) or indirectly (external) through a ConfigMap. A custom ConfigMap must be placed under the log4j.properties or log4j2.properties key. For the Kafka Connect log4j.rootLogger logger, you can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.
14: Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).
15: Prometheus metrics, which are enabled by referencing a ConfigMap containing configuration for the Prometheus JMX exporter in this example. You can enable metrics without further configuration using a reference to a ConfigMap containing an empty file under metricsConfig.valueFrom.configMapKeyRef.key.
16: JVM configuration options to optimize performance for the Virtual Machine (VM) running Kafka Connect.
17: ADVANCED OPTION: Container image configuration, which is recommended only in special situations.
18: SPECIALIZED OPTION: Rack awareness configuration for the deployment. This is a specialized option intended for a deployment within the same location, not across regions. Use this option if you want connectors to consume from the closest replica rather than the leader replica. In certain cases, consuming from the closest replica can improve network utilization or reduce costs . The topologyKey must match a node label containing the rack ID. The example used in this configuration specifies a zone using the standard topology.kubernetes.io/zone label. To consume from the closest replica, enable the RackAwareReplicaSelector in the Kafka broker configuration.
19: Template customization. Here a pod is scheduled with anti-affinity, so the pod is not scheduled on nodes with the same hostname.
20: Environment variables are set for distributed tracing.

Create or update the resource:
```
oc apply -f KAFKA-CONNECT-CONFIG-FILE
```
If authorization is enabled for Kafka Connect, configure Kafka Connect users to enable access to the Kafka Connect consumer group and topics.

Additional resources

Introducing distributed tracing

2.3.2. Kafka Connect configuration for multiple instances

If you are running multiple instances of Kafka Connect, you have to change the default configuration of the following config properties:

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: The Kafka Connect cluster ID within Kafka.
2: Kafka topic that stores connector offsets.
3: Kafka topic that stores connector and task status configurations.
4: Kafka topic that stores connector and task status updates.

Note

Values for the three topics must be the same for all Kafka Connect instances with the same group.id.

Unless you change the default settings, each Kafka Connect instance connecting to the same Kafka cluster is deployed with the same values. What happens, in effect, is all instances are coupled to run in a cluster and use the same topics.

If multiple Kafka Connect clusters try to use the same topics, Kafka Connect will not work as expected and generate errors.

If you wish to run multiple Kafka Connect instances, change the values of these properties for each instance.

2.3.3. Configuring Kafka Connect user authorization

This procedure describes how to authorize user access to Kafka Connect.

When any type of authorization is being used in Kafka, a Kafka Connect user requires read/write access rights to the consumer group and the internal topics of Kafka Connect.

The properties for the consumer group and internal topics are automatically configured by AMQ Streams, or they can be specified explicitly in the spec of the KafkaConnect resource.

Example configuration properties in the KafkaConnect resource

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect
spec:
  # ...
  config:
    group.id: my-connect-cluster 1
    offset.storage.topic: my-connect-cluster-offsets 2
    config.storage.topic: my-connect-cluster-configs 3
    status.storage.topic: my-connect-cluster-status 4
    # ...
  # ...

1: The Kafka Connect cluster ID within Kafka.
2: Kafka topic that stores connector offsets.
3: Kafka topic that stores connector and task status configurations.
4: Kafka topic that stores connector and task status updates.

This procedure shows how access is provided when simple authorization is being used.

Simple authorization uses ACL rules, handled by the Kafka AclAuthorizer plugin, to provide the right level of access. For more information on configuring a KafkaUser resource to use simple authorization, see the AclRule schema reference.

Note

The default values for the consumer group and topics will differ when running multiple instances.

Prerequisites

An OpenShift cluster
A running Cluster Operator

Procedure

Edit the authorization property in the KafkaUser resource to provide access rights to the user.

In the following example, access rights are configured for the Kafka Connect topics and consumer group using literal name values:

Property	Name
`offset.storage.topic`	`connect-cluster-offsets`
`status.storage.topic`	`connect-cluster-status`
`config.storage.topic`	`connect-cluster-configs`
`group`	`connect-cluster`

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  # ...
  authorization:
    type: simple
    acls:
      # access to offset.storage.topic
      - resource:
          type: topic
          name: connect-cluster-offsets
          patternType: literal
        operations:
          - Create
          - Describe
          - Read
          - Write
        host: "*"
      # access to status.storage.topic
      - resource:
          type: topic
          name: connect-cluster-status
          patternType: literal
        operations:
          - Create
          - Describe
          - Read
          - Write
        host: "*"
      # access to config.storage.topic
      - resource:
          type: topic
          name: connect-cluster-configs
          patternType: literal
        operations:
          - Create
          - Describe
          - Read
          - Write
        host: "*"
      # consumer group
      - resource:
          type: group
          name: connect-cluster
          patternType: literal
        operations:
          - Read
        host: "*"

Create or update the resource.
```
oc apply -f KAFKA-USER-CONFIG-FILE
```

2.3.4. List of Kafka Connect cluster resources

The following resources are created by the Cluster Operator in the OpenShift cluster:

connect-cluster-name-connect: Deployment which is in charge to create the Kafka Connect worker node pods.
connect-cluster-name-connect-api: Service which exposes the REST interface for managing the Kafka Connect cluster.
connect-cluster-name-config: ConfigMap which contains the Kafka Connect ancillary configuration and is mounted as a volume by the Kafka broker pods.
connect-cluster-name-connect: Pod Disruption Budget configured for the Kafka Connect worker nodes.

2.3.5. Integrating with the Red Hat build of Debezium for change data capture

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:

Data replication
Updating caches and search indexes
Simplifying monolithic applications
Data integration
Enabling streaming queries

To capture database changes, deploy Kafka Connect with a Debezium database connector. You configure a KafkaConnector resource to define the connector instance.

For more information on deploying the Red Hat build of Debezium with AMQ Streams, refer to the product documentation. The documentation includes a Getting Started with Debezium guide that guides you through the process of setting up the services and connector required to view change event records for database updates.

2.4. Kafka MirrorMaker 2.0 cluster configuration

Configure a Kafka MirrorMaker 2.0 deployment using the KafkaMirrorMaker2 resource. MirrorMaker 2.0 replicates data between two or more Kafka clusters, within or across data centers.

Section 12.2.126, “KafkaMirrorMaker2 schema reference” describes the full schema of the KafkaMirrorMaker2 resource.

MirrorMaker 2.0 resource configuration differs from the previous version of MirrorMaker. If you choose to use MirrorMaker 2.0, there is currently no legacy support, so any resources must be manually converted into the new format.

2.4.1. MirrorMaker 2.0 data replication

Data replication across clusters supports scenarios that require:

Recovery of data in the event of a system failure
Aggregation of data for analysis
Restriction of data access to a specific cluster
Provision of data at a specific location to improve latency

2.4.1.1. MirrorMaker 2.0 configuration

MirrorMaker 2.0 consumes messages from a source Kafka cluster and writes them to a target Kafka cluster.

MirrorMaker 2.0 uses:

Source cluster configuration to consume data from the source cluster
Target cluster configuration to output data to the target cluster

MirrorMaker 2.0 is based on the Kafka Connect framework, connectors managing the transfer of data between clusters.

You configure MirrorMaker 2.0 to define the Kafka Connect deployment, including the connection details of the source and target clusters, and then run a set of MirrorMaker 2.0 connectors to make the connection.

MirrorMaker 2.0 consists of the following connectors:

MirrorSourceConnector: The source connector replicates topics from a source cluster to a target cluster. It also replicates ACLs and is necessary for the MirrorCheckpointConnector to run.
MirrorCheckpointConnector: The checkpoint connector periodically tracks offsets. If enabled, it also synchronizes consumer group offsets between the source and target cluster.
MirrorHeartbeatConnector: The heartbeat connector periodically checks connectivity between the source and target cluster.

Note

If you are using the User Operator to manage ACLs, ACL replication through the connector is not possible.

The process of mirroring data from a source cluster to a target cluster is asynchronous. Each MirrorMaker 2.0 instance mirrors data from one source cluster to one target cluster. You can use more than one MirrorMaker 2.0 instance to mirror data between any number of clusters.

Figure 2.1. Replication across two clusters

By default, a check for new topics in the source cluster is made every 10 minutes. You can change the frequency by adding refresh.topics.interval.seconds to the source connector configuration.

2.4.1.1.1. Cluster configuration

You can use MirrorMaker 2.0 in active/passive or active/active cluster configurations.

active/active cluster configuration: An active/active configuration has two active clusters replicating data bidirectionally. Applications can use either cluster. Each cluster can provide the same data. In this way, you can make the same data available in different geographical locations. As consumer groups are active in both clusters, consumer offsets for replicated topics are not synchronized back to the source cluster.
active/passive cluster configuration: An active/passive configuration has an active cluster replicating data to a passive cluster. The passive cluster remains on standby. You might use the passive cluster for data recovery in the event of system failure.

The expectation is that producers and consumers connect to active clusters only. A MirrorMaker 2.0 cluster is required at each target destination.

2.4.1.1.2. Bidirectional replication (active/active)

The MirrorMaker 2.0 architecture supports bidirectional replication in an active/active cluster configuration.

Each cluster replicates the data of the other cluster using the concept of source and remote topics. As the same topics are stored in each cluster, remote topics are automatically renamed by MirrorMaker 2.0 to represent the source cluster. The name of the originating cluster is prepended to the name of the topic.

Figure 2.2. Topic renaming

MirrorMaker 2.0 bidirectional architecture

By flagging the originating cluster, topics are not replicated back to that cluster.

The concept of replication through remote topics is useful when configuring an architecture that requires data aggregation. Consumers can subscribe to source and remote topics within the same cluster, without the need for a separate aggregation cluster.

2.4.1.1.3. Unidirectional replication (active/passive)

The MirrorMaker 2.0 architecture supports unidirectional replication in an active/passive cluster configuration.

You can use an active/passive cluster configuration to make backups or migrate data to another cluster. In this situation, you might not want automatic renaming of remote topics.

You can override automatic renaming by adding IdentityReplicationPolicy to the source connector configuration. With this configuration applied, topics retain their original names.

2.4.1.2. Topic configuration synchronization

MirrorMaker 2.0 supports topic configuration synchronization between source and target clusters. You specify source topics in the MirrorMaker 2.0 configuration. MirrorMaker 2.0 monitors the source topics. MirrorMaker 2.0 detects and propagates changes to the source topics to the remote topics. Changes might include automatically creating missing topics and partitions.

Note

In most cases you write to local topics and read from remote topics. Though write operations are not prevented on remote topics, they should be avoided.

2.4.1.3. Offset tracking

MirrorMaker 2.0 tracks offsets for consumer groups using internal topics.

offset-syncs topic: The offset-syncs topic maps the source and target offsets for replicated topic partitions from record metadata.
checkpoints topic: The checkpoints topic maps the last committed offset in the source and target cluster for replicated topic partitions in each consumer group.

As they used internally by MirrorMaker 2.0, you do not interact directly with these topics.

MirrorCheckpointConnector emits checkpoints for offset tracking. Offsets for the checkpoints topic are tracked at predetermined intervals through configuration. Both topics enable replication to be fully restored from the correct offset position on failover.

The location of the offset-syncs topic is the source cluster by default. You can use the offset-syncs.topic.location connector configuration to change this to the target cluster. You need read/write access to the cluster that contains the topic. Using the target cluster as the location of the offset-syncs topic allows you to use MirrorMaker 2.0 even if you have only read access to the source cluster.

2.4.1.4. Synchronizing consumer group offsets

The __consumer_offsets topic stores information on committed offsets for each consumer group. Offset synchronization periodically transfers the consumer offsets for the consumer groups of a source cluster into the consumer offsets topic of a target cluster.

Offset synchronization is particularly useful in an active/passive configuration. If the active cluster goes down, consumer applications can switch to the passive (standby) cluster and pick up from the last transferred offset position.

To use topic offset synchronization, enable the synchronization by adding sync.group.offsets.enabled to the checkpoint connector configuration, and setting the property to true. Synchronization is disabled by default.

When using the IdentityReplicationPolicy in the source connector, it also has to be configured in the checkpoint connector configuration. This ensures that the mirrored consumer offsets will be applied for the correct topics.

Consumer offsets are only synchronized for consumer groups that are not active in the target cluster. If the consumer groups are in the target cluster, the synchronization cannot be performed and an UNKNOWN_MEMBER_ID error is returned.

If enabled, the synchronization of offsets from the source cluster is made periodically. You can change the frequency by adding sync.group.offsets.interval.seconds and emit.checkpoints.interval.seconds to the checkpoint connector configuration. The properties specify the frequency in seconds that the consumer group offsets are synchronized, and the frequency of checkpoints emitted for offset tracking. The default for both properties is 60 seconds. You can also change the frequency of checks for new consumer groups using the refresh.groups.interval.seconds property, which is performed every 10 minutes by default.

Because the synchronization is time-based, any switchover by consumers to a passive cluster will likely result in some duplication of messages.

Note

If you have an application written in Java, you can use the RemoteClusterUtils.java utility to synchronize offsets through the application. The utility fetches remote offsets for a consumer group from the checkpoints topic.

2.4.1.5. Connectivity checks

MirrorHeartbeatConnector emits heartbeats to check connectivity between clusters.

An internal heartbeat topic is replicated from the source cluster. Target clusters use the heartbeat topic to check the following:

The connector managing connectivity between clusters is running
The source cluster is available

2.4.2. Connector configuration

Use Mirrormaker 2.0 connector configuration for the internal connectors that orchestrate the synchronization of data between Kafka clusters.

The following table describes connector properties and the connectors you configure to use them.

Table 2.1. MirrorMaker 2.0 connector configuration properties
Property	sourceConnector	checkpointConnector	heartbeatConnector
admin.timeout.ms Timeout for admin tasks, such as detecting new topics. Default is `60000` (1 minute).	✓	✓	✓
replication.policy.class Policy to define the remote topic naming convention. Default is `org.apache.kafka.connect.mirror.DefaultReplicationPolicy`.	✓	✓	✓
replication.policy.separator The separator used for topic naming in the target cluster. Default is `.` (dot). It is only used when the `replication.policy.class` is the `DefaultReplicationPolicy`.	✓	✓	✓
consumer.poll.timeout.ms Timeout when polling the source cluster. Default is `1000` (1 second).	✓	✓
offset-syncs.topic.location The location of the `offset-syncs` topic, which can be the `source` (default) or `target` cluster.	✓	✓
topic.filter.class Topic filter to select the topics to replicate. Default is `org.apache.kafka.connect.mirror.DefaultTopicFilter`.	✓	✓
config.property.filter.class Topic filter to select the topic configuration properties to replicate. Default is `org.apache.kafka.connect.mirror.DefaultConfigPropertyFilter`.	✓
config.properties.exclude Topic configuration properties that should not be replicated. Supports comma-separated property names and regular expressions.	✓
offset.lag.max Maximum allowable (out-of-sync) offset lag before a remote partition is synchronized. Default is `100`.	✓
offset-syncs.topic.replication.factor Replication factor for the internal `offset-syncs` topic. Default is `3`.	✓
refresh.topics.enabled Enables check for new topics and partitions. Default is `true`.	✓
refresh.topics.interval.seconds Frequency of topic refresh. Default is `600` (10 minutes).	✓
replication.factor The replication factor for new topics. Default is `2`.	✓
sync.topic.acls.enabled Enables synchronization of ACLs from the source cluster. Default is `true`. Not compatible with the User Operator.	✓
sync.topic.acls.interval.seconds Frequency of ACL synchronization. Default is `600` (10 minutes).	✓
sync.topic.configs.enabled Enables synchronization of topic configuration from the source cluster. Default is `true`.	✓
sync.topic.configs.interval.seconds Frequency of topic configuration synchronization. Default `600` (10 minutes).	✓
checkpoints.topic.replication.factor Replication factor for the internal `checkpoints` topic. Default is `3`.		✓
emit.checkpoints.enabled Enables synchronization of consumer offsets to the target cluster. Default is `true`.		✓
emit.checkpoints.interval.seconds Frequency of consumer offset synchronization. Default is `60` (1 minute).		✓
group.filter.class Group filter to select the consumer groups to replicate. Default is `org.apache.kafka.connect.mirror.DefaultGroupFilter`.		✓
refresh.groups.enabled Enables check for new consumer groups. Default is `true`.		✓
refresh.groups.interval.seconds Frequency of consumer group refresh. Default is `600` (10 minutes).		✓
sync.group.offsets.enabled Enables synchronization of consumer group offsets to the target cluster `__consumer_offsets` topic. Default is `false`.		✓
sync.group.offsets.interval.seconds Frequency of consumer group offset synchronization. Default is `60` (1 minute).		✓
emit.heartbeats.enabled Enables connectivity checks on the target cluster. Default is `true`.			✓
emit.heartbeats.interval.seconds Frequency of connectivity checks. Default is `1` (1 second).			✓
heartbeats.topic.replication.factor Replication factor for the internal `heartbeats` topic. Default is `3`.			✓

2.4.3. Connector producer and consumer configuration

MirrorMaker 2.0 connectors use internal producers and consumers. If needed, you can configure these producers and consumers to override the default settings.

For example, you can increase the batch.size for the source producer that sends topics to the target Kafka cluster to better accommodate large volumes of messages.

Important

Producer and consumer configuration options depend on the MirrorMaker 2.0 implementation, and may be subject to change.

The following tables describe the producers and consumers for each of the connectors and where you can add configuration.

Table 2.2. Source connector producers and consumers
Type	Description	Configuration
Producer	Sends topic messages to the target Kafka cluster. Consider tuning the configuration of this producer when it is handling large volumes of data.	`mirrors.sourceConnector.config: producer.override.*`
Producer	Writes to the `offset-syncs` topic, which maps the source and target offsets for replicated topic partitions.	`mirrors.sourceConnector.config: producer.*`
Consumer	Retrieves topic messages from the source Kafka cluster.	`mirrors.sourceConnector.config: consumer.*`

Table 2.3. Checkpoint connector producers and consumers
Type	Description	Configuration
Producer	Emits consumer offset checkpoints.	`mirrors.checkpointConnector.config: producer.override.*`
Consumer	Loads the `offset-syncs` topic.	`mirrors.checkpointConnector.config: consumer.*`

Note

You can set offset-syncs.topic.location to target to use the target Kafka cluster as the location of the offset-syncs topic.

Table 2.4. Heartbeat connector producer
Type	Description	Configuration
Producer	Emits heartbeats.	`mirrors.heartbeatConnector.config: producer.override.*`

The following example shows how you configure the producers and consumers.

Example configuration for connector producers and consumers

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker2
spec:
  version: 3.3.1
  # ...
  mirrors:
  - sourceCluster: "my-cluster-source"
    targetCluster: "my-cluster-target"
    sourceConnector:
      tasksMax: 5
      config:
        producer.override.batch.size: 327680
        producer.override.linger.ms: 100
        producer.request.timeout.ms: 30000
        consumer.fetch.max.bytes: 52428800
        # ...
    checkpointConnector:
      config:
        producer.override.request.timeout.ms: 30000
        consumer.max.poll.interval.ms: 300000
        # ...
    heartbeatConnector:
      config:
        producer.override.request.timeout.ms: 30000
        # ...

Additional resources

Section 12.2.130, “KafkaMirrorMaker2ConnectorSpec schema reference”
Section 12.2.129, “KafkaMirrorMaker2MirrorSpec schema reference”

2.4.4. Specifying a maximum number of tasks

Connectors create the tasks that are responsible for moving data in and out of Kafka. Each connector comprises one or more tasks that are distributed across a group of worker pods that run the tasks. Increasing the number of tasks can help with performance issues when replicating a large number of partitions or synchronizing the offsets of a large number of consumer groups.

Tasks run in parallel. Workers are assigned one or more tasks. A single task is handled by one worker pod, so you don’t need more worker pods than tasks. If there are more tasks than workers, workers handle multiple tasks.

You can specify the maximum number of connector tasks in your MirrorMaker configuration using the tasksMax property. Without specifying a maximum number of tasks, the default setting is a single task.

The heartbeat connector always uses a single task.

The number of tasks that are started for the source and checkpoint connectors is the lower value between the maximum number of possible tasks and the value for tasksMax. For the source connector, the maximum number of tasks possible is one for each partition being replicated from the source cluster. For the checkpoint connector, the maximum number of tasks possible is one for each consumer group being replicated from the source cluster. When setting a maximum number of tasks, consider the number of partitions and the hardware resources that support the process.

If the infrastructure supports the processing overhead, increasing the number of tasks can improve throughput and latency. For example, adding more tasks reduces the time taken to poll the source cluster when there is a high number of partitions or consumer groups.

Increasing the number of tasks for the checkpoint connector is useful when you have a large number of partitions.

Increasing the number of tasks for the source connector

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker2
spec:
  # ...
  mirrors:
  - sourceCluster: "my-cluster-source"
    targetCluster: "my-cluster-target"
    sourceConnector:
      tasksMax: 10
  # ...

Increasing the number of tasks for the checkpoint connector is useful when you have a large number of consumer groups.

Increasing the number of tasks for the checkpoint connector

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker2
spec:
  # ...
  mirrors:
  - sourceCluster: "my-cluster-source"
    targetCluster: "my-cluster-target"
    checkpointConnector:
      tasksMax: 10
  # ...

By default, MirrorMaker 2.0 checks for new consumer groups every 10 minutes. You can adjust the refresh.groups.interval.seconds configuration to change the frequency. Take care when adjusting lower. More frequent checks can have a negative impact on performance.

2.4.4.1. Checking connector task operations

If you are using Prometheus and Grafana to monitor your deployment, you can check MirrorMaker 2.0 performance. The example MirrorMaker 2.0 Grafana dashboard provided with AMQ Streams shows the following metrics related to tasks and latency.

The number of tasks
Replication latency
Offset synchronization latency

Additional resources

Grafana dashboards

2.4.5. ACL rules synchronization

ACL access to remote topics is possible if you are not using the User Operator.

If AclAuthorizer is being used, without the User Operator, ACL rules that manage access to brokers also apply to remote topics. Users that can read a source topic can read its remote equivalent.

Note

OAuth 2.0 authorization does not support access to remote topics in this way.

2.4.6. Configuring Kafka MirrorMaker 2.0

Use the properties of the KafkaMirrorMaker2 resource to configure your Kafka MirrorMaker 2.0 deployment. Use MirrorMaker 2.0 to synchronize data between Kafka clusters.

The configuration must specify:

Each Kafka cluster
Connection information for each cluster, including authentication
The replication flow and direction
- Cluster to cluster
- Topic to topic

Note

The previous version of MirrorMaker continues to be supported. If you wish to use the resources configured for the previous version, they must be updated to the format supported by MirrorMaker 2.0.

MirrorMaker 2.0 provides default configuration values for properties such as replication factors. A minimal configuration, with defaults left unchanged, would be something like this example:

Minimal configuration for MirrorMaker 2.0

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker2
spec:
  version: 3.3.1
  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: {}

You can configure access control for source and target clusters using mTLS or SASL authentication. This procedure shows a configuration that uses TLS encryption and mTLS authentication for the source and target cluster.

You can specify the topics and consumer groups you wish to replicate from a source cluster in the KafkaMirrorMaker2 resource. You use the topicsPattern and groupsPattern properties to do this. You can provide a list of names or use a regular expression. By default, all topics and consumer groups are replicated if you do not set the topicsPattern and groupsPattern properties. You can also replicate all topics and consumer groups by using ".*" as a regular expression. However, try to specify only the topics and consumer groups you need to avoid causing any unnecessary extra load on the cluster.

Handling high volumes of messages

You can tune the configuration to handle high volumes of messages. For more information, see Section 2.7, “Handling high volumes of messages”.

Prerequisites

AMQ Streams is running
Source and target Kafka clusters are available

Procedure

Edit the spec properties for the KafkaMirrorMaker2 resource.
The properties you can configure are shown in this example configuration:
```
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker2
spec:
  version: 3.3.1 1
  replicas: 3 2
  connectCluster: "my-cluster-target" 3
  clusters: 4
  - alias: "my-cluster-source" 5
    authentication: 6
      certificateAndKey:
        certificate: source.crt
        key: source.key
        secretName: my-user-source
      type: tls
    bootstrapServers: my-cluster-source-kafka-bootstrap:9092 7
    tls: 8
      trustedCertificates:
      - certificate: ca.crt
        secretName: my-cluster-source-cluster-ca-cert
  - alias: "my-cluster-target" 9
    authentication: 10
      certificateAndKey:
        certificate: target.crt
        key: target.key
        secretName: my-user-target
      type: tls
    bootstrapServers: my-cluster-target-kafka-bootstrap:9092 11
    config: 12
      config.storage.replication.factor: 1
      offset.storage.replication.factor: 1
      status.storage.replication.factor: 1
      ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 13
      ssl.enabled.protocols: "TLSv1.2"
      ssl.protocol: "TLSv1.2"
      ssl.endpoint.identification.algorithm: HTTPS 14
    tls: 15
      trustedCertificates:
      - certificate: ca.crt
        secretName: my-cluster-target-cluster-ca-cert
  mirrors: 16
  - sourceCluster: "my-cluster-source" 17
    targetCluster: "my-cluster-target" 18
    sourceConnector: 19
      tasksMax: 10 20
      config:
        replication.factor: 1 21
        offset-syncs.topic.replication.factor: 1 22
        sync.topic.acls.enabled: "false" 23
        refresh.topics.interval.seconds: 60 24
        replication.policy.separator: "" 25
        replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy" 26
    heartbeatConnector: 27
      config:
        heartbeats.topic.replication.factor: 1 28
    checkpointConnector: 29
      config:
        checkpoints.topic.replication.factor: 1 30
        refresh.groups.interval.seconds: 600 31
        sync.group.offsets.enabled: true 32
        sync.group.offsets.interval.seconds: 60 33
        emit.checkpoints.interval.seconds: 60 34
        replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
    topicsPattern: "topic1|topic2|topic3" 35
    groupsPattern: "group1|group2|group3" 36
  resources: 37
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  logging: 38
    type: inline
    loggers:
      connect.root.logger.level: "INFO"
  readinessProbe: 39
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  jvmOptions: 40
    "-Xmx": "1g"
    "-Xms": "1g"
  image: my-org/my-image:latest 41
  rack:
    topologyKey: topology.kubernetes.io/zone 42
  template: 43
    pod:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: application
                    operator: In
                    values:
                      - postgresql
                      - mongodb
              topologyKey: "kubernetes.io/hostname"
    connectContainer: 44
      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 45
  externalConfiguration: 46
    env:
      - name: AWS_ACCESS_KEY_ID
        valueFrom:
          secretKeyRef:
            name: aws-creds
            key: awsAccessKey
      - name: AWS_SECRET_ACCESS_KEY
        valueFrom:
          secretKeyRef:
            name: aws-creds
            key: awsSecretAccessKey
```
1
The Kafka Connect and Mirror Maker 2.0 version, which will always be the same.
2
The number of replica nodes for the workers that run tasks.
3
Kafka cluster alias for Kafka Connect, which must specify the target Kafka cluster. The Kafka cluster is used by Kafka Connect for its internal topics.
4
Specification for the Kafka clusters being synchronized.
5
Cluster alias for the source Kafka cluster.
6
Authentication for the source cluster, specified as mTLS, token-based OAuth, SASL-based SCRAM-SHA-256/SCRAM-SHA-512, or PLAIN.
7
Bootstrap server for connection to the source Kafka cluster.
8
TLS encryption with key names under which TLS certificates are stored in X.509 format for the source Kafka cluster. If certificates are stored in the same secret, it can be listed multiple times.
9
Cluster alias for the target Kafka cluster.
10
Authentication for the target Kafka cluster is configured in the same way as for the source Kafka cluster.
11
Bootstrap server for connection to the target Kafka cluster.
12
Kafka Connect configuration. Standard Apache Kafka configuration may be provided, restricted to those properties not managed directly by AMQ Streams.
13
SSL properties for external listeners to run with a specific cipher suite for a TLS version.
14
Hostname verification is enabled by setting to HTTPS. An empty string disables the verification.
15
TLS encryption for the target Kafka cluster is configured in the same way as for the source Kafka cluster.
16
MirrorMaker 2.0 connectors.
17
Cluster alias for the source cluster used by the MirrorMaker 2.0 connectors.
18
Cluster alias for the target cluster used by the MirrorMaker 2.0 connectors.
19
Configuration for the MirrorSourceConnector that creates remote topics. The config overrides the default configuration options.
20
The maximum number of tasks that the connector may create. Tasks handle the data replication and run in parallel. If the infrastructure supports the processing overhead, increasing this value can improve throughput. Kafka Connect distributes the tasks between members of the cluster. If there are more tasks than workers, workers are assigned multiple tasks. For sink connectors, aim to have one task for each topic partition consumed. For source connectors, the number of tasks that can run in parallel may also depend on the external system. The connector creates fewer than the maximum number of tasks if it cannot achieve the parallelism.
21
Replication factor for mirrored topics created at the target cluster.
22
Replication factor for the MirrorSourceConnector offset-syncs internal topic that maps the offsets of the source and target clusters.
23
When ACL rules synchronization is enabled, ACLs are applied to synchronized topics. The default is true. This feature is not compatible with the User Operator. If you are using the User Operator, set this property to false.
24
Optional setting to change the frequency of checks for new topics. The default is for a check every 10 minutes.
25
Defines the separator used for the renaming of remote topics.
26
Adds a policy that overrides 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. To configure topic offset synchronization, this property must also be set for the checkpointConnector.config.
27
Configuration for the MirrorHeartbeatConnector that performs connectivity checks. The config overrides the default configuration options.
28
Replication factor for the heartbeat topic created at the target cluster.
29
Configuration for the MirrorCheckpointConnector that tracks offsets. The config overrides the default configuration options.
30
Replication factor for the checkpoints topic created at the target cluster.
31
Optional setting to change the frequency of checks for new consumer groups. The default is for a check every 10 minutes.
32
Optional setting to synchronize consumer group offsets, which is useful for recovery in an active/passive configuration. Synchronization is not enabled by default.
33
If the synchronization of consumer group offsets is enabled, you can adjust the frequency of the synchronization.
34
Adjusts the frequency of checks for offset tracking. If you change the frequency of offset synchronization, you might also need to adjust the frequency of these checks.
35
Topic replication from the source cluster defined as a comma-separated list or regular expression pattern. The source connector replicates the specified topics. The checkpoint connector tracks offsets for the specified topics. Here we request three topics by name.
36
Consumer group replication from the source cluster defined as a comma-separated list or regular expression pattern. The checkpoint connector replicates the specified consumer groups. Here we request three consumer groups by name.
37
Requests for reservation of supported resources, currently cpu and memory, and limits to specify the maximum resources that can be consumed.
38
Specified Kafka Connect loggers and log levels added directly (inline) or indirectly (external) through a ConfigMap. A custom ConfigMap must be placed under the log4j.properties or log4j2.properties key. For the Kafka Connect log4j.rootLogger logger, you can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.
39
Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).
40
JVM configuration options to optimize performance for the Virtual Machine (VM) running Kafka MirrorMaker.
41
ADVANCED OPTION: Container image configuration, which is recommended only in special situations.
42
SPECIALIZED OPTION: Rack awareness configuration for the deployment. This is a specialized option intended for a deployment within the same location, not across regions. Use this option if you want connectors to consume from the closest replica rather than the leader replica. In certain cases, consuming from the closest replica can improve network utilization or reduce costs . The topologyKey must match a node label containing the rack ID. The example used in this configuration specifies a zone using the standard topology.kubernetes.io/zone label. To consume from the closest replica, enable the RackAwareReplicaSelector in the Kafka broker configuration.
43
Template customization. Here a pod is scheduled with anti-affinity, so the pod is not scheduled on nodes with the same hostname.
44
Environment variables are set for distributed tracing.
45
Distributed tracing is enabled for Jaeger.
46
External configuration for an OpenShift Secret mounted to Kafka MirrorMaker as an environment variable. You can also use configuration provider plugins to load configuration values from external sources.

Create or update the resource:

oc apply -f MIRRORMAKER-CONFIGURATION-FILE

Additional resources

Introducing distributed tracing

2.4.7. Securing a Kafka MirrorMaker 2.0 deployment

This procedure describes in outline the configuration required to secure a MirrorMaker 2.0 deployment.

You need separate configuration for the source Kafka cluster and the target Kafka cluster. You also need separate user configuration to provide the credentials required for MirrorMaker to connect to the source and target Kafka clusters.

For the Kafka clusters, you specify internal listeners for secure connections within an OpenShift cluster and external listeners for connections outside the OpenShift cluster.

You can configure authentication and authorization mechanisms. The security options implemented for the source and target Kafka clusters must be compatible with the security options implemented for MirrorMaker 2.0.

After you have created the cluster and user authentication credentials, you specify them in your MirrorMaker configuration for secure connections.

Note

In this procedure, the certificates generated by the Cluster Operator are used, but you can replace them by installing your own certificates. You can also configure your listener to use a Kafka listener certificate managed by an external CA (certificate authority.

Before you start

Before starting this procedure, take a look at the example configuration files provided by AMQ Streams. They include examples for securing a deployment of MirrorMaker 2.0 using mTLS or SCRAM-SHA-512 authentication. The examples specify internal listeners for connecting within an OpenShift cluster.

The examples provide the configuration for full authorization, including all the ACLs needed by MirrorMaker 2.0 to allow operations on the source and target Kafka clusters.

Prerequisites

AMQ Streams is running
Separate namespaces for source and target clusters

The procedure assumes that the source and target Kafka clusters are installed to separate namespaces If you want to use the Topic Operator, you’ll need to do this. The Topic Operator only watches a single cluster in a specified namespace.

By separating the clusters into namespaces, you will need to copy the cluster secrets so they can be accessed outside the namespace. You need to reference the secrets in the MirrorMaker configuration.

Procedure

Configure two Kafka resources, one to secure the source Kafka cluster and one to secure the target Kafka cluster.

You can add listener configuration for authentication and enable authorization.

In this example, an internal listener is configured for a Kafka cluster with TLS encryption and mTLS authentication. Kafka simple authorization is enabled.

Example source Kafka cluster configuration with TLS encryption and mTLS authentication

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-source-cluster
spec:
  kafka:
    version: 3.3.1
    replicas: 1
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
    authorization:
      type: simple
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      min.insync.replicas: 1
      inter.broker.protocol.version: "3.3"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
  zookeeper:
    replicas: 1
    storage:
      type: persistent-claim
      size: 100Gi
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}

Example target Kafka cluster configuration with TLS encryption and mTLS authentication

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-target-cluster
spec:
  kafka:
    version: 3.3.1
    replicas: 1
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
    authorization:
      type: simple
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      min.insync.replicas: 1
      inter.broker.protocol.version: "3.3"
    storage:
      type: jbod
      volumes:
        - id: 0
          type: persistent-claim
          size: 100Gi
          deleteClaim: false
  zookeeper:
    replicas: 1
    storage:
      type: persistent-claim
      size: 100Gi
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}

Create or update the Kafka resources in separate namespaces.
```
oc apply -f <kafka_configuration_file> -n <namespace>
```
The Cluster Operator creates the listeners and sets up the cluster and client certificate authority (CA) certificates to enable authentication within the Kafka cluster.
The certificates are created in the secret <cluster_name>-cluster-ca-cert.

Configure two KafkaUser resources, one for a user of the source Kafka cluster and one for a user of the target Kafka cluster.

Configure the same authentication and authorization types as the corresponding source and target Kafka cluster. For example, if you used tls authentication and the simple authorization type in the Kafka configuration for the source Kafka cluster, use the same in the KafkaUser configuration.
Configure the ACLs needed by MirrorMaker 2.0 to allow operations on the source and target Kafka clusters.
The ACLs are used by the internal MirrorMaker connectors, and by the underlying Kafka Connect framework.

Example source user configuration for mTLS authentication

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-source-user
  labels:
    strimzi.io/cluster: my-source-cluster
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
      # MirrorSourceConnector
      - resource: # Not needed if offset-syncs.topic.location=target
          type: topic
          name: mm2-offset-syncs.my-target-cluster.internal
        operations:
          - Create
          - DescribeConfigs
          - Read
          - Write
      - resource: # Needed for every topic which is mirrored
          type: topic
          name: "*"
        operations:
          - DescribeConfigs
          - Read
      # MirrorCheckpointConnector
      - resource:
          type: cluster
        operations:
          - Describe
      - resource: # Needed for every group for which offsets are synced
          type: group
          name: "*"
        operations:
          - Describe
      - resource: # Not needed if offset-syncs.topic.location=target
          type: topic
          name: mm2-offset-syncs.my-target-cluster.internal
        operations:
          - Read

Example target user configuration for mTLS authentication

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-target-user
  labels:
    strimzi.io/cluster: my-target-cluster
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
      # Underlying Kafka Connect internal topics to store configuration, offsets, or status
      - resource:
          type: group
          name: mirrormaker2-cluster
        operations:
          - Read
      - resource:
          type: topic
          name: mirrormaker2-cluster-configs
        operations:
          - Create
          - Describe
          - DescribeConfigs
          - Read
          - Write
      - resource:
          type: topic
          name: mirrormaker2-cluster-status
        operations:
          - Create
          - Describe
          - DescribeConfigs
          - Read
          - Write
      - resource:
          type: topic
          name: mirrormaker2-cluster-offsets
        operations:
          - Create
          - Describe
          - DescribeConfigs
          - Read
          - Write
      # MirrorSourceConnector
      - resource: # Needed for every topic which is mirrored
          type: topic
          name: "*"
        operations:
          - Create
          - Alter
          - AlterConfigs
          - Write
      # MirrorCheckpointConnector
      - resource:
          type: cluster
        operations:
          - Describe
      - resource:
          type: topic
          name: my-source-cluster.checkpoints.internal
        operations:
          - Create
          - Describe
          - Read
          - Write
      - resource: # Needed for every group for which the offset is synced
          type: group
          name: "*"
        operations:
          - Read
          - Describe
      # MirrorHeartbeatConnector
      - resource:
          type: topic
          name: heartbeats
        operations:
          - Create
          - Describe
          - Write

Note

You can use a certificate issued outside the User Operator by setting type to tls-external. For more information, see User authentication.

Create or update a KafkaUser resource in each of the namespaces you created for the source and target Kafka clusters.
```
oc apply -f <kafka_user_configuration_file> -n <namespace>
```
The User Operator creates the users representing the client (MirrorMaker), and the security credentials used for client authentication, based on the chosen authentication type.
The User Operator creates a new secret with the same name as the KafkaUser resource. The secret contains a private and public key for mTLS authentication. The public key is contained in a user certificate, which is signed by the clients CA.

Configure a KafkaMirrorMaker2 resource with the authentication details to connect to the source and target Kafka clusters.

Example MirrorMaker 2.0 configuration with TLS encryption and mTLS authentication

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
spec:
  version: 3.3.1
  replicas: 1
  connectCluster: "my-target-cluster"
  clusters:
    - alias: "my-source-cluster"
      bootstrapServers: my-source-cluster-kafka-bootstrap:9093
      tls: 1
        trustedCertificates:
          - secretName: my-source-cluster-cluster-ca-cert
            certificate: ca.crt
      authentication: 2
        type: tls
        certificateAndKey:
          secretName: my-source-user
          certificate: user.crt
          key: user.key
    - alias: "my-target-cluster"
      bootstrapServers: my-target-cluster-kafka-bootstrap:9093
      tls: 3
        trustedCertificates:
          - secretName: my-target-cluster-cluster-ca-cert
            certificate: ca.crt
      authentication: 4
        type: tls
        certificateAndKey:
          secretName: my-target-user
          certificate: user.crt
          key: user.key
      config:
        # -1 means it will use the default replication factor configured in the broker
        config.storage.replication.factor: -1
        offset.storage.replication.factor: -1
        status.storage.replication.factor: -1
  mirrors:
    - sourceCluster: "my-source-cluster"
      targetCluster: "my-target-cluster"
      sourceConnector:
        config:
          replication.factor: 1
          offset-syncs.topic.replication.factor: 1
          sync.topic.acls.enabled: "false"
      heartbeatConnector:
        config:
          heartbeats.topic.replication.factor: 1
      checkpointConnector:
        config:
          checkpoints.topic.replication.factor: 1
          sync.group.offsets.enabled: "true"
      topicsPattern: "topic1|topic2|topic3"
      groupsPattern: "group1|group2|group3"

1: The TLS certificates for the source Kafka cluster. If they are in a separate namespace, copy the cluster secrets from the namespace of the Kafka cluster.
2: The user authentication for accessing the source Kafka cluster using the TLS mechanism.
3: The TLS certificates for the target Kafka cluster.
4: The user authentication for accessing the target Kafka cluster.

Create or update the KafkaMirrorMaker2 resource in the same namespace as the target Kafka cluster.
```
oc apply -f <mirrormaker2_configuration_file> -n <namespace_of_target_cluster>
```

Additional resources

Supported listener authentication options
Supported authorization options for a Kafka cluster
Securing Kafka brokers
Securing user access to Kafka
Managing TLS certificates

2.4.8. Performing a restart of a Kafka MirrorMaker 2.0 connector

This procedure describes how to manually trigger a restart of a Kafka MirrorMaker 2.0 connector by using an OpenShift annotation.

Prerequisites

The Cluster Operator is running.

Procedure

Find the name of the KafkaMirrorMaker2 custom resource that controls the Kafka MirrorMaker 2.0 connector you want to restart:
```
oc get KafkaMirrorMaker2
```
Find the name of the Kafka MirrorMaker 2.0 connector to be restarted from the KafkaMirrorMaker2 custom resource.
```
oc describe KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME
```
To restart the connector, annotate the KafkaMirrorMaker2 resource in OpenShift. In this example, oc annotate restarts a connector named my-source->my-target.MirrorSourceConnector:
```
oc annotate KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME "strimzi.io/restart-connector=my-source->my-target.MirrorSourceConnector"
```
Wait for the next reconciliation to occur (every two minutes by default).
The Kafka MirrorMaker 2.0 connector is restarted, as long as the annotation was detected by the reconciliation process. When the restart request is accepted, the annotation is removed from the KafkaMirrorMaker2 custom resource.

Additional resources

Kafka MirrorMaker 2.0 cluster configuration.

2.4.9. Performing a restart of a Kafka MirrorMaker 2.0 connector task

This procedure describes how to manually trigger a restart of a Kafka MirrorMaker 2.0 connector task by using an OpenShift annotation.

Prerequisites

The Cluster Operator is running.

Procedure

Find the name of the KafkaMirrorMaker2 custom resource that controls the Kafka MirrorMaker 2.0 connector you want to restart:
```
oc get KafkaMirrorMaker2
```
Find the name of the Kafka MirrorMaker 2.0 connector and the ID of the task to be restarted from the KafkaMirrorMaker2 custom resource. Task IDs are non-negative integers, starting from 0.
```
oc describe KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME
```
To restart the connector task, annotate the KafkaMirrorMaker2 resource in OpenShift. In this example, oc annotate restarts task 0 of a connector named my-source->my-target.MirrorSourceConnector:
```
oc annotate KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME "strimzi.io/restart-connector-task=my-source->my-target.MirrorSourceConnector:0"
```
Wait for the next reconciliation to occur (every two minutes by default).
The Kafka MirrorMaker 2.0 connector task is restarted, as long as the annotation was detected by the reconciliation process. When the restart task request is accepted, the annotation is removed from the KafkaMirrorMaker2 custom resource.

Additional resources

Kafka MirrorMaker 2.0 cluster configuration.

2.5. Kafka MirrorMaker cluster configuration

Configure a Kafka MirrorMaker deployment using the KafkaMirrorMaker resource. KafkaMirrorMaker replicates data between Kafka clusters.

Section 12.2.108, “KafkaMirrorMaker schema reference” describes the full schema of the KafkaMirrorMaker resource.

You can use AMQ Streams with MirrorMaker or MirrorMaker 2.0. MirrorMaker 2.0 is the latest version, and offers a more efficient way to mirror data between Kafka clusters.

Important

Kafka MirrorMaker 1 (referred to as just MirrorMaker in the documentation) has been deprecated in Apache Kafka 3.0.0 and will be removed in Apache Kafka 4.0.0. As a result, the KafkaMirrorMaker custom resource which is used to deploy Kafka MirrorMaker 1 has been deprecated in AMQ Streams as well. The KafkaMirrorMaker resource will be removed from AMQ Streams when we adopt Apache Kafka 4.0.0. As a replacement, use the KafkaMirrorMaker2 custom resource with the IdentityReplicationPolicy.

2.5.1. Configuring Kafka MirrorMaker

Use the properties of the KafkaMirrorMaker resource to configure your Kafka MirrorMaker deployment.

You can configure access control for producers and consumers using TLS or SASL authentication. This procedure shows a configuration that uses TLS encryption and mTLS authentication on the consumer and producer side.

Prerequisites

See the Deploying and Upgrading AMQ Streams on OpenShift guide for instructions on running a:
- Cluster Operator
- Kafka cluster
Source and target Kafka clusters must be available

Procedure

Edit the spec properties for the KafkaMirrorMaker resource.

The properties you can configure are shown in this example configuration:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker
metadata:
  name: my-mirror-maker
spec:
  replicas: 3 1
  consumer:
    bootstrapServers: my-source-cluster-kafka-bootstrap:9092 2
    groupId: "my-group" 3
    numStreams: 2 4
    offsetCommitInterval: 120000 5
    tls: 6
      trustedCertificates:
      - secretName: my-source-cluster-ca-cert
        certificate: ca.crt
    authentication: 7
      type: tls
      certificateAndKey:
        secretName: my-source-secret
        certificate: public.crt
        key: private.key
    config: 8
      max.poll.records: 100
      receive.buffer.bytes: 32768
      ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 9
      ssl.enabled.protocols: "TLSv1.2"
      ssl.protocol: "TLSv1.2"
      ssl.endpoint.identification.algorithm: HTTPS 10
  producer:
    bootstrapServers: my-target-cluster-kafka-bootstrap:9092
    abortOnSendFailure: false 11
    tls:
      trustedCertificates:
      - secretName: my-target-cluster-ca-cert
        certificate: ca.crt
    authentication:
      type: tls
      certificateAndKey:
        secretName: my-target-secret
        certificate: public.crt
        key: private.key
    config:
      compression.type: gzip
      batch.size: 8192
      ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 12
      ssl.enabled.protocols: "TLSv1.2"
      ssl.protocol: "TLSv1.2"
      ssl.endpoint.identification.algorithm: HTTPS 13
  include: "my-topic|other-topic" 14
  resources: 15
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  logging: 16
    type: inline
    loggers:
      mirrormaker.root.logger: "INFO"
  readinessProbe: 17
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  metricsConfig: 18
   type: jmxPrometheusExporter
   valueFrom:
     configMapKeyRef:
       name: my-config-map
       key: my-key
  jvmOptions: 19
    "-Xmx": "1g"
    "-Xms": "1g"
  image: my-org/my-image:latest 20
  template: 21
    pod:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: application
                    operator: In
                    values:
                      - postgresql
                      - mongodb
              topologyKey: "kubernetes.io/hostname"
    connectContainer: 22
      env:
        - name: JAEGER_SERVICE_NAME
          value: my-jaeger-service
        - name: JAEGER_AGENT_HOST
          value: jaeger-agent-name
        - name: JAEGER_AGENT_PORT
          value: "6831"
  tracing: 23
    type: jaeger

1: The number of replica nodes.
2: Bootstrap servers for consumer and producer.
3: Group ID for the consumer.
4: The number of consumer streams.
5: The offset auto-commit interval in milliseconds.
6: TLS encryption with key names under which TLS certificates are stored in X.509 format for consumer or producer. If certificates are stored in the same secret, it can be listed multiple times.
7: Authentication for consumer or producer, specified as mTLS, token-based OAuth, SASL-based SCRAM-SHA-256/SCRAM-SHA-512, or PLAIN.
8: Kafka configuration options for consumer and producer.
9: SSL properties for external listeners to run with a specific cipher suite for a TLS version.
10: Hostname verification is enabled by setting to HTTPS. An empty string disables the verification.
11: If the abortOnSendFailure property is set to true, Kafka MirrorMaker will exit and the container will restart following a send failure for a message.
12: SSL properties for external listeners to run with a specific cipher suite for a TLS version.
13: Hostname verification is enabled by setting to HTTPS. An empty string disables the verification.
14: A included topics mirrored from source to target Kafka cluster.
15: Requests for reservation of supported resources, currently cpu and memory, and limits to specify the maximum resources that can be consumed.
16: Specified loggers and log levels added directly (inline) or indirectly (external) through a ConfigMap. A custom ConfigMap must be placed under the log4j.properties or log4j2.properties key. MirrorMaker has a single logger called mirrormaker.root.logger. You can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.
17: Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).
18: Prometheus metrics, which are enabled by referencing a ConfigMap containing configuration for the Prometheus JMX exporter in this example. You can enable metrics without further configuration using a reference to a ConfigMap containing an empty file under metricsConfig.valueFrom.configMapKeyRef.key.
19: JVM configuration options to optimize performance for the Virtual Machine (VM) running Kafka MirrorMaker.
20: ADVANCED OPTION: Container image configuration, which is recommended only in special situations.
21: Template customization. Here a pod is scheduled with anti-affinity, so the pod is not scheduled on nodes with the same hostname.
22: Environment variables are set for distributed tracing.
23: Distributed tracing is enabled for Jaeger.

Warning

With the abortOnSendFailure property set to false, the producer attempts to send the next message in a topic. The original message might be lost, as there is no attempt to resend a failed message.

Create or update the resource:
```
oc apply -f <your-file>
```

Additional resources

Introducing distributed tracing

2.5.2. List of Kafka MirrorMaker cluster resources

The following resources are created by the Cluster Operator in the OpenShift cluster:

<mirror-maker-name>-mirror-maker: Deployment which is responsible for creating the Kafka MirrorMaker pods.
<mirror-maker-name>-config: ConfigMap which contains ancillary configuration for the Kafka MirrorMaker, and is mounted as a volume by the Kafka broker pods.
<mirror-maker-name>-mirror-maker: Pod Disruption Budget configured for the Kafka MirrorMaker worker nodes.

2.6. Kafka Bridge cluster configuration

Configure a Kafka Bridge deployment using the KafkaBridge resource. Kafka Bridge provides an API for integrating HTTP-based clients with a Kafka cluster.

Section 12.2.114, “KafkaBridge schema reference” describes the full schema of the KafkaBridge resource.

2.6.1. Configuring the Kafka Bridge

Use the Kafka Bridge to make HTTP-based requests to the Kafka cluster.

Use the properties of the KafkaBridge resource to configure your Kafka Bridge deployment.

In order to prevent issues arising when client consumer requests are processed by different Kafka Bridge instances, address-based routing must be employed to ensure that requests are routed to the right Kafka Bridge instance. Additionally, each independent Kafka Bridge instance must have a replica. A Kafka Bridge instance has its own state which is not shared with another instances.

Prerequisites

An OpenShift cluster
A running Cluster Operator

See the Deploying and Upgrading AMQ Streams on OpenShift guide for instructions on running a:

Procedure

Edit the spec properties for the KafkaBridge resource.

The properties you can configure are shown in this example configuration:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaBridge
metadata:
  name: my-bridge
spec:
  replicas: 3 1
  bootstrapServers: <cluster_name>-cluster-kafka-bootstrap:9092 2
  tls: 3
    trustedCertificates:
      - secretName: my-cluster-cluster-cert
        certificate: ca.crt
      - secretName: my-cluster-cluster-cert
        certificate: ca2.crt
  authentication: 4
    type: tls
    certificateAndKey:
      secretName: my-secret
      certificate: public.crt
      key: private.key
  http: 5
    port: 8080
    cors: 6
      allowedOrigins: "https://strimzi.io"
      allowedMethods: "GET,POST,PUT,DELETE,OPTIONS,PATCH"
  consumer: 7
    config:
      auto.offset.reset: earliest
  producer: 8
    config:
      delivery.timeout.ms: 300000
  resources: 9
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  logging: 10
    type: inline
    loggers:
      logger.bridge.level: "INFO"
      # enabling DEBUG just for send operation
      logger.send.name: "http.openapi.operation.send"
      logger.send.level: "DEBUG"
  jvmOptions: 11
    "-Xmx": "1g"
    "-Xms": "1g"
  readinessProbe: 12
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  image: my-org/my-image:latest 13
  template: 14
    pod:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: application
                    operator: In
                    values:
                      - postgresql
                      - mongodb
              topologyKey: "kubernetes.io/hostname"
    bridgeContainer: 15
      env:
        - name: JAEGER_SERVICE_NAME
          value: my-jaeger-service
        - name: JAEGER_AGENT_HOST
          value: jaeger-agent-name
        - name: JAEGER_AGENT_PORT
          value: "6831"

1: The number of replica nodes.
2: Bootstrap server for connection to the target Kafka cluster. Use the name of the Kafka cluster as the <cluster_name>.
3: TLS encryption with key names under which TLS certificates are stored in X.509 format for the source Kafka cluster. If certificates are stored in the same secret, it can be listed multiple times.
4: Authentication for the Kafka Bridge cluster, specified as mTLS, token-based OAuth, SASL-based SCRAM-SHA-256/SCRAM-SHA-512, or PLAIN. By default, the Kafka Bridge connects to Kafka brokers without authentication.
5: HTTP access to Kafka brokers.
6: CORS access specifying selected resources and access methods. Additional HTTP headers in requests describe the origins that are permitted access to the Kafka cluster.
7: Consumer configuration options.
8: Producer configuration options.
9: Requests for reservation of supported resources, currently cpu and memory, and limits to specify the maximum resources that can be consumed.
10: Specified Kafka Bridge loggers and log levels added directly (inline) or indirectly (external) through a ConfigMap. A custom ConfigMap must be placed under the log4j.properties or log4j2.properties key. For the Kafka Bridge loggers, you can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.
11: JVM configuration options to optimize performance for the Virtual Machine (VM) running the Kafka Bridge.
12: Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).
13: Optional: Container image configuration, which is recommended only in special situations.
14: Template customization. Here a pod is scheduled with anti-affinity, so the pod is not scheduled on nodes with the same hostname.
15: Environment variables are set for distributed tracing.

Create or update the resource:
```
oc apply -f KAFKA-BRIDGE-CONFIG-FILE
```

Additional resources

2.6.2. List of Kafka Bridge cluster resources

The following resources are created by the Cluster Operator in the OpenShift cluster:

bridge-cluster-name-bridge: Deployment which is in charge to create the Kafka Bridge worker node pods.
bridge-cluster-name-bridge-service: Service which exposes the REST interface of the Kafka Bridge cluster.
bridge-cluster-name-bridge-config: ConfigMap which contains the Kafka Bridge ancillary configuration and is mounted as a volume by the Kafka broker pods.
bridge-cluster-name-bridge: Pod Disruption Budget configured for the Kafka Bridge worker nodes.

2.7. Handling high volumes of messages

If your AMQ Streams deployment needs to handle a high volume of messages, you can use configuration options to optimize for throughput and latency.

Producer and consumer configuration can help control the size and frequency of requests to Kafka brokers. For more information on the configuration options, see the following:

You can also use the same configuration options with the producers and consumers used by the Kafka Connect runtime source connectors (including MirrorMaker 2.0) and sink connectors.

Source connectors

Producers from the Kafka Connect runtime send messages to the Kafka cluster.
For MirrorMaker 2.0, since the source system is Kafka, consumers retrieve messages from a source Kafka cluster.

Sink connectors

Consumers from the Kafka Connect runtime retrieve messages from the Kafka cluster.

For consumers, you might increase the amount of data fetched in a single fetch request to reduce latency. You increase the fetch request size using the fetch.max.bytes and max.partition.fetch.bytes properties. You can also set a maximum limit on the number of messages returned from the consumer buffer using the max.poll.records property.

For MirrorMaker 2.0, configure the fetch.max.bytes, max.partition.fetch.bytes, and max.poll.records values at the source connector level (consumer.*), as they relate to the specific consumer that fetches messages from the source.

For producers, you might increase the size of the message batches sent in a single produce request. You increase the batch size using the batch.size property. A larger batch size reduces the number of outstanding messages ready to be sent and the size of the backlog in the message queue. Messages being sent to the same partition are batched together. A produce request is sent to the target cluster when the batch size is reached. By increasing the batch size, produce requests are delayed and more messages are added to the batch and sent to brokers at the same time. This can improve throughput when you have just a few topic partitions that handle large numbers of messages.

Consider the number and size of the records that the producer handles for a suitable producer batch size.

Use linger.ms to add a wait time in milliseconds to delay produce requests when producer load decreases. The delay means that more records can be added to batches if they are under the maximum batch size.

Configure the batch.size and linger.ms values at the source connector level (producer.override.*), as they relate to the specific producer that sends messages to the target Kafka cluster.

For Kafka Connect source connectors, the data streaming pipeline to the target Kafka cluster is as follows:

Data streaming pipeline for Kafka Connect source connector

external data source → (Kafka Connect tasks) source message queue → producer buffer → target Kafka topic

For Kafka Connect sink connectors, the data streaming pipeline to the target external data source is as follows:

Data streaming pipeline for Kafka Connect sink connector

source Kafka topic → (Kafka Connect tasks) sink message queue → consumer buffer → external data source

For MirrorMaker 2.0, the data mirroring pipeline to the target Kafka cluster is as follows:

Data mirroring pipeline for MirrorMaker 2.0

source Kafka topic → (Kafka Connect tasks) source message queue → producer buffer → target Kafka topic

The producer sends messages in its buffer to topics in the target Kafka cluster. While this is happening, Kafka Connect tasks continue to poll the data source to add messages to the source message queue.

The size of the producer buffer for the source connector is set using the producer.override.buffer.memory property. Tasks wait for a specified timeout period (offset.flush.timeout.ms) before the buffer is flushed. This should be enough time for the sent messages to be acknowledged by the brokers and offset data committed. The source task does not wait for the producer to empty the message queue before committing offsets, except during shutdown.

If the producer is unable to keep up with the throughput of messages in the source message queue, buffering is blocked until there is space available in the buffer within a time period bounded by max.block.ms. Any unacknowledged messages still in the buffer are sent during this period. New messages are not added to the buffer until these messages are acknowledged and flushed.

You can try the following configuration changes to keep the underlying source message queue of outstanding messages at a manageable size:

Increasing the default value in milliseconds of the offset.flush.timeout.ms
Ensuring that there are enough CPU and memory resources
Increasing the number of tasks that run in parallel by doing the following:
- Increasing the number of tasks that run in parallel using the tasksMax property
- Increasing the number of worker nodes that run tasks using the replicas property

Consider the number of tasks that can run in parallel according to the available CPU and memory resources and number of worker nodes. You might need to keep adjusting the configuration values until they have the desired effect.

2.7.1. Configuring Kafka Connect for high-volume messages

Kafka Connect fetches data from the source external data system and hands it to the Kafka Connect runtime producers so that it’s replicated to the target cluster.

The following example shows configuration for Kafka Connect using the KafkaConnect custom resource.

Example Kafka Connect configuration for handling high volumes of messages

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  replicas: 3
  config:
    offset.flush.timeout.ms: 10000
    # ...
  resources:
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  # ...

Producer configuration is added for the source connector, which is managed using the KafkaConnector custom resource.

Example source connector configuration for handling high volumes of messages

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
  config:
    producer.override.batch.size: 327680
    producer.override.linger.ms: 100
    # ...

Note

FileStreamSourceConnector and FileStreamSinkConnector are provided as example connectors. For information on deploying them as KafkaConnector resources, see Deploying example KafkaConnector resources.

Consumer configuration is added for the sink connector.

Example sink connector configuration for handling high volumes of messages

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-sink-connector
  labels:
    strimzi.io/cluster: my-connect-cluster
spec:
  class: org.apache.kafka.connect.file.FileStreamSinkConnector
  tasksMax: 2
  config:
    consumer.fetch.max.bytes: 52428800
    consumer.max.partition.fetch.bytes: 1048576
    consumer.max.poll.records: 500
    # ...

If you are using the Kafka Connect API instead of the KafkaConnector custom resource to manage your connectors, you can add the connector configuration as a JSON object.

Example curl request to add source connector configuration for handling high volumes of messages

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"
      "producer.override.batch.size": 327680
      "producer.override.linger.ms": 100
    }
}'

2.7.2. Configuring MirrorMaker 2.0 for high-volume messages

MirrorMaker 2.0 fetches data from the source cluster and hands it to the Kafka Connect runtime producers so that it’s replicated to the target cluster.

The following example shows the configuration for MirrorMaker 2.0 using the KafkaMirrorMaker2 custom resource.

Example MirrorMaker 2.0 configuration for handling high volumes of messages

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker2
spec:
  version: 3.3.1
  replicas: 1
  connectCluster: "my-cluster-target"
  clusters:
  - alias: "my-cluster-source"
    bootstrapServers: my-cluster-source-kafka-bootstrap:9092
  - alias: "my-cluster-target"
    config:
      offset.flush.timeout.ms: 10000
    bootstrapServers: my-cluster-target-kafka-bootstrap:9092
  mirrors:
  - sourceCluster: "my-cluster-source"
    targetCluster: "my-cluster-target"
    sourceConnector:
      tasksMax: 2
      config:
        producer.override.batch.size: 327680
        producer.override.linger.ms: 100
        consumer.fetch.max.bytes: 52428800
        consumer.max.partition.fetch.bytes: 1048576
        consumer.max.poll.records: 500
    # ...
  resources:
    requests:
      cpu: "1"
      memory: Gi
    limits:
      cpu: "2"
      memory: 4Gi

2.7.3. Checking the MirrorMaker 2.0 message flow

If you are using Prometheus and Grafana to monitor your deployment, you can check the MirrorMaker 2.0 message flow.

The example MirrorMaker 2.0 Grafana dashboards provided with AMQ Streams show the following metrics related to the flush pipeline.

The number of messages in Kafka Connect’s outstanding messages queue
The available bytes of the producer buffer
The offset commit timeout in milliseconds

You can use these metrics to gauge whether or not you need to tune your configuration based on the volume of messages.

Additional resources

2.8. Customizing OpenShift resources

An AMQ Streams deployment creates OpenShift resources, such as Deployments, StatefulSets, Pods, and Services. These resources are managed by AMQ Streams operators. Only the operator that is responsible for managing a particular OpenShift resource can change that resource. If you try to manually change an operator-managed OpenShift resource, the operator will revert your changes back.

Changing an operator-managed OpenShift resource can be useful if you want to perform certain tasks, such as:

Adding custom labels or annotations that control how Pods are treated by Istio or other services
Managing how Loadbalancer-type Services are created by the cluster

You can make the changes using the template property in the AMQ Streams custom resources. The template property is supported in the following resources. The API reference provides more details about the customizable fields.

Kafka.spec.kafka: See Section 12.2.33, “KafkaClusterTemplate schema reference”
Kafka.spec.zookeeper: See Section 12.2.43, “ZookeeperClusterTemplate schema reference”
Kafka.spec.entityOperator: See Section 12.2.48, “EntityOperatorTemplate schema reference”
Kafka.spec.kafkaExporter: See Section 12.2.55, “KafkaExporterTemplate schema reference”
Kafka.spec.cruiseControl: See Section 12.2.51, “CruiseControlTemplate schema reference”
KafkaConnect.spec: See Section 12.2.71, “KafkaConnectTemplate schema reference”
KafkaMirrorMaker.spec: See Section 12.2.112, “KafkaMirrorMakerTemplate schema reference”
KafkaMirrorMaker2.spec: See Section 12.2.71, “KafkaConnectTemplate schema reference”
KafkaBridge.spec: See Section 12.2.121, “KafkaBridgeTemplate schema reference”
KafkaUser.spec: See Section 12.2.106, “KafkaUserTemplate schema reference”

In the following example, the template property is used to modify the labels in a Kafka broker’s pod.

Example template customization

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  labels:
    app: my-cluster
spec:
  kafka:
    # ...
    template:
      pod:
        metadata:
          labels:
            mylabel: myvalue
    # ...

2.8.1. Customizing the image pull policy

AMQ Streams allows you to customize the image pull policy for containers in all pods deployed by the Cluster Operator. The image pull policy is configured using the environment variable STRIMZI_IMAGE_PULL_POLICY in the Cluster Operator deployment. The STRIMZI_IMAGE_PULL_POLICY environment variable can be set to three different values:

Always: Container images are pulled from the registry every time the pod is started or restarted.
IfNotPresent: Container images are pulled from the registry only when they were not pulled before.
Never: Container images are never pulled from the registry.

The image pull policy can be currently customized only for all Kafka, Kafka Connect, and Kafka MirrorMaker clusters at once. Changing the policy will result in a rolling update of all your Kafka, Kafka Connect, and Kafka MirrorMaker clusters.

Additional resources

For more information about Cluster Operator configuration, see Section 7.2, “Using the Cluster Operator”.
For more information about Image Pull Policies, see Disruptions.

2.8.2. Applying a termination grace period

Apply a termination grace period to give a Kafka cluster enough time to shut down cleanly.

Specify the time using the terminationGracePeriodSeconds property. Add the property to the template.pod configuration of the Kafka custom resource.

The time you add will depend on the size of your Kafka cluster. The OpenShift default for the termination grace period is 30 seconds. If you observe that your clusters are not shutting down cleanly, you can increase the termination grace period.

A termination grace period is applied every time a pod is restarted. The period begins when OpenShift sends a term (termination) signal to the processes running in the pod. The period should reflect the amount of time required to transfer the processes of the terminating pod to another pod before they are stopped. After the period ends, a kill signal stops any processes still running in the pod.

The following example adds a termination grace period of 120 seconds to the Kafka custom resource. You can also specify the configuration in the custom resources of other Kafka components.

Example termination grace period configuration

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    template:
      pod:
        terminationGracePeriodSeconds: 120
        # ...
    # ...

2.9. Configuring pod scheduling

When two applications are scheduled to the same OpenShift node, both applications might use the same resources like disk I/O and impact performance. That can lead to performance degradation. Scheduling Kafka pods in a way that avoids sharing nodes with other critical workloads, using the right nodes or dedicated a set of nodes only for Kafka are the best ways how to avoid such problems.

2.9.1. Specifying affinity, tolerations, and topology spread constraints

Use affinity, tolerations and topology spread constraints to schedule the pods of kafka resources onto nodes. Affinity, tolerations and topology spread constraints are configured using the affinity, tolerations, and topologySpreadConstraint properties in following resources:

Kafka.spec.kafka.template.pod
Kafka.spec.zookeeper.template.pod
Kafka.spec.entityOperator.template.pod
KafkaConnect.spec.template.pod
KafkaBridge.spec.template.pod
KafkaMirrorMaker.spec.template.pod
KafkaMirrorMaker2.spec.template.pod

The format of the affinity, tolerations, and topologySpreadConstraint properties follows the OpenShift specification. The affinity configuration can include different types of affinity:

Pod affinity and anti-affinity
Node affinity

Additional resources

2.9.1.1. Use pod anti-affinity to avoid critical applications sharing nodes

Use pod anti-affinity to ensure that critical applications are never scheduled on the same disk. When running a Kafka cluster, it is recommended to use pod anti-affinity to ensure that the Kafka brokers do not share nodes with other workloads, such as databases.

2.9.1.2. Use node affinity to schedule workloads onto specific nodes

The OpenShift cluster usually consists of many different types of worker nodes. Some are optimized for CPU heavy workloads, some for memory, while other might be optimized for storage (fast local SSDs) or network. Using different nodes helps to optimize both costs and performance. To achieve the best possible performance, it is important to allow scheduling of AMQ Streams components to use the right nodes.

OpenShift uses node affinity to schedule workloads onto specific nodes. Node affinity allows you to create a scheduling constraint for the node on which the pod will be scheduled. The constraint is specified as a label selector. You can specify the label using either the built-in node label like beta.kubernetes.io/instance-type or custom labels to select the right node.

2.9.1.3. Use node affinity and tolerations for dedicated nodes

Use taints to create dedicated nodes, then schedule Kafka pods on the dedicated nodes by configuring node affinity and tolerations.

Cluster administrators can mark selected OpenShift nodes as tainted. Nodes with taints are excluded from regular scheduling and normal pods will not be scheduled to run on them. Only services which can tolerate the taint set on the node can be scheduled on it. The only other services running on such nodes will be system services such as log collectors or software defined networks.

Running Kafka and its components on dedicated nodes can have many advantages. There will be no other applications running on the same nodes which could cause disturbance or consume the resources needed for Kafka. That can lead to improved performance and stability.

2.9.2. Configuring pod anti-affinity to schedule each Kafka broker on a different worker node

Many Kafka brokers or ZooKeeper nodes can run on the same OpenShift worker node. If the worker node fails, they will all become unavailable at the same time. To improve reliability, you can use podAntiAffinity configuration to schedule each Kafka broker or ZooKeeper node on a different OpenShift worker node.

Prerequisites

An OpenShift cluster
A running Cluster Operator

Procedure

Edit the affinity property in the resource specifying the cluster deployment. To make sure that no worker nodes are shared by Kafka brokers or ZooKeeper nodes, use the strimzi.io/name label. Set the topologyKey to kubernetes.io/hostname to specify that the selected pods are not scheduled on nodes with the same hostname. This will still allow the same worker node to be shared by a single Kafka broker and a single ZooKeeper node. For example:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    # ...
    template:
      pod:
        affinity:
          podAntiAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              - labelSelector:
                  matchExpressions:
                    - key: strimzi.io/name
                      operator: In
                      values:
                        - CLUSTER-NAME-kafka
                topologyKey: "kubernetes.io/hostname"
    # ...
  zookeeper:
    # ...
    template:
      pod:
        affinity:
          podAntiAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              - labelSelector:
                  matchExpressions:
                    - key: strimzi.io/name
                      operator: In
                      values:
                        - CLUSTER-NAME-zookeeper
                topologyKey: "kubernetes.io/hostname"
    # ...

Where CLUSTER-NAME is the name of your Kafka custom resource.

If you even want to make sure that a Kafka broker and ZooKeeper node do not share the same worker node, use the strimzi.io/cluster label. For example:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    # ...
    template:
      pod:
        affinity:
          podAntiAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              - labelSelector:
                  matchExpressions:
                    - key: strimzi.io/cluster
                      operator: In
                      values:
                        - CLUSTER-NAME
                topologyKey: "kubernetes.io/hostname"
    # ...
  zookeeper:
    # ...
    template:
      pod:
        affinity:
          podAntiAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              - labelSelector:
                  matchExpressions:
                    - key: strimzi.io/cluster
                      operator: In
                      values:
                        - CLUSTER-NAME
                topologyKey: "kubernetes.io/hostname"
    # ...

Where CLUSTER-NAME is the name of your Kafka custom resource.

Create or update the resource.
```
oc apply -f <kafka_configuration_file>
```

2.9.3. Configuring pod anti-affinity in Kafka components

Pod anti-affinity configuration helps with the stability and performance of Kafka brokers. By using podAntiAffinity, OpenShift will not schedule Kafka brokers on the same nodes as other workloads. Typically, you want to avoid Kafka running on the same worker node as other network or storage intensive applications such as databases, storage or other messaging platforms.

Prerequisites

An OpenShift cluster
A running Cluster Operator

Procedure

Edit the affinity property in the resource specifying the cluster deployment. Use labels to specify the pods which should not be scheduled on the same nodes. The topologyKey should be set to kubernetes.io/hostname to specify that the selected pods should not be scheduled on nodes with the same hostname. For example:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    # ...
    template:
      pod:
        affinity:
          podAntiAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              - labelSelector:
                  matchExpressions:
                    - key: application
                      operator: In
                      values:
                        - postgresql
                        - mongodb
                topologyKey: "kubernetes.io/hostname"
    # ...
  zookeeper:
    # ...

Create or update the resource.
This can be done using oc apply:
```
oc apply -f <kafka_configuration_file>
```

2.9.4. Configuring node affinity in Kafka components

Prerequisites

An OpenShift cluster
A running Cluster Operator

Procedure

Label the nodes where AMQ Streams components should be scheduled.
This can be done using oc label:
```
oc label node NAME-OF-NODE node-type=fast-network
```
Alternatively, some of the existing labels might be reused.

Edit the affinity property in the resource specifying the cluster deployment. For example:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    # ...
    template:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
                - matchExpressions:
                  - key: node-type
                    operator: In
                    values:
                    - fast-network
    # ...
  zookeeper:
    # ...

Create or update the resource.
This can be done using oc apply:
```
oc apply -f <kafka_configuration_file>
```

2.9.5. Setting up dedicated nodes and scheduling pods on them

Prerequisites

An OpenShift cluster
A running Cluster Operator

Procedure

Select the nodes which should be used as dedicated.
Make sure there are no workloads scheduled on these nodes.
Set the taints on the selected nodes:
This can be done using oc adm taint:
```
oc adm taint node NAME-OF-NODE dedicated=Kafka:NoSchedule
```
Additionally, add a label to the selected nodes as well.
This can be done using oc label:
```
oc label node NAME-OF-NODE dedicated=Kafka
```

Edit the affinity and tolerations properties in the resource specifying the cluster deployment.

For example:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    # ...
    template:
      pod:
        tolerations:
          - key: "dedicated"
            operator: "Equal"
            value: "Kafka"
            effect: "NoSchedule"
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
              - matchExpressions:
                - key: dedicated
                  operator: In
                  values:
                  - Kafka
    # ...
  zookeeper:
    # ...

Create or update the resource.
This can be done using oc apply:
```
oc apply -f <kafka_configuration_file>
```

2.10. Logging configuration

Configure logging levels in the custom resources of Kafka components and AMQ Streams Operators. You can specify the logging levels directly in the spec.logging property of the custom resource. Or you can define the logging properties in a ConfigMap that’s referenced in the custom resource using the configMapKeyRef property.

The advantages of using a ConfigMap are that the logging properties are maintained in one place and are accessible to more than one resource. You can also reuse the ConfigMap for more than one resource. If you are using a ConfigMap to specify loggers for AMQ Streams Operators, you can also append the logging specification to add filters.

You specify a logging type in your logging specification:

inline when specifying logging levels directly
external when referencing a ConfigMap

Example inline logging configuration

spec:
  # ...
  logging:
    type: inline
    loggers:
      kafka.root.logger.level: "INFO"

Example external logging configuration

spec:
  # ...
  logging:
    type: external
    valueFrom:
      configMapKeyRef:
        name: my-config-map
        key: my-config-map-key

Values for the name and key of the ConfigMap are mandatory. Default logging is used if the name or key is not set.

2.10.1. Logging options for Kafka components and operators

For more information on configuring logging for specific Kafka components or operators, see the following sections.

Kafka component logging

Kafka logging
ZooKeeper logging
Kafka Connect and Mirror Maker 2.0 logging
MirrorMaker logging
Kafka Bridge logging
Cruise Control logging

Operator logging

Cluster Operator logging
Topic Operator logging
User Operator logging

2.10.2. Creating a ConfigMap for logging

To use a ConfigMap to define logging properties, you create the ConfigMap and then reference it as part of the logging definition in the spec of a resource.

The ConfigMap must contain the appropriate logging configuration.

log4j.properties for Kafka components, ZooKeeper, and the Kafka Bridge
log4j2.properties for the Topic Operator and User Operator

The configuration must be placed under these properties.

In this procedure a ConfigMap defines a root logger for a Kafka resource.

Procedure

Create the ConfigMap.
You can create the ConfigMap as a YAML file or from a properties file.
ConfigMap example with a root logger definition for Kafka:
```
kind: ConfigMap
apiVersion: v1
metadata:
  name: logging-configmap
data:
  log4j.properties:
    kafka.root.logger.level="INFO"
```
If you are using a properties file, specify the file at the command line:
```
oc create configmap logging-configmap --from-file=log4j.properties
```
The properties file defines the logging configuration:
```
# Define the logger
kafka.root.logger.level="INFO"
# ...
```
Define external logging in the spec of the resource, setting the logging.valueFrom.configMapKeyRef.name to the name of the ConfigMap and logging.valueFrom.configMapKeyRef.key to the key in this ConfigMap.
```
spec:
  # ...
  logging:
    type: external
    valueFrom:
      configMapKeyRef:
        name: logging-configmap
        key: log4j.properties
```
Create or update the resource.
```
oc apply -f <kafka_configuration_file>
```

2.10.3. Adding logging filters to Operators

If you are using a ConfigMap to configure the (log4j2) logging levels for AMQ Streams Operators, you can also define logging filters to limit what’s returned in the log.

Logging filters are useful when you have a large number of logging messages. Suppose you set the log level for the logger as DEBUG (rootLogger.level="DEBUG"). Logging filters reduce the number of logs returned for the logger at that level, so you can focus on a specific resource. When the filter is set, only log messages matching the filter are logged.

Filters use markers to specify what to include in the log. You specify a kind, namespace and name for the marker. For example, if a Kafka cluster is failing, you can isolate the logs by specifying the kind as Kafka, and use the namespace and name of the failing cluster.

This example shows a marker filter for a Kafka cluster named my-kafka-cluster.

Basic logging filter configuration

rootLogger.level="INFO"
appender.console.filter.filter1.type=MarkerFilter 1
appender.console.filter.filter1.onMatch=ACCEPT 2
appender.console.filter.filter1.onMismatch=DENY 3
appender.console.filter.filter1.marker=Kafka(my-namespace/my-kafka-cluster) 4

1: The MarkerFilter type compares a specified marker for filtering.
2: The onMatch property accepts the log if the marker matches.
3: The onMismatch property rejects the log if the marker does not match.
4: The marker used for filtering is in the format KIND(NAMESPACE/NAME-OF-RESOURCE).

You can create one or more filters. Here, the log is filtered for two Kafka clusters.

Multiple logging filter configuration

appender.console.filter.filter1.type=MarkerFilter
appender.console.filter.filter1.onMatch=ACCEPT
appender.console.filter.filter1.onMismatch=DENY
appender.console.filter.filter1.marker=Kafka(my-namespace/my-kafka-cluster-1)
appender.console.filter.filter2.type=MarkerFilter
appender.console.filter.filter2.onMatch=ACCEPT
appender.console.filter.filter2.onMismatch=DENY
appender.console.filter.filter2.marker=Kafka(my-namespace/my-kafka-cluster-2)

Adding filters to the Cluster Operator

To add filters to the Cluster Operator, update its logging ConfigMap YAML file (install/cluster-operator/050-ConfigMap-strimzi-cluster-operator.yaml).

Procedure

Update the 050-ConfigMap-strimzi-cluster-operator.yaml file to add the filter properties to the ConfigMap.

In this example, the filter properties return logs only for the my-kafka-cluster Kafka cluster:

kind: ConfigMap
apiVersion: v1
metadata:
  name: strimzi-cluster-operator
data:
  log4j2.properties:
    #...
    appender.console.filter.filter1.type=MarkerFilter
    appender.console.filter.filter1.onMatch=ACCEPT
    appender.console.filter.filter1.onMismatch=DENY
    appender.console.filter.filter1.marker=Kafka(my-namespace/my-kafka-cluster)

Alternatively, edit the ConfigMap directly:

oc edit configmap strimzi-cluster-operator

If you updated the YAML file instead of editing the ConfigMap directly, apply the changes by deploying the ConfigMap:
```
oc create -f install/cluster-operator/050-ConfigMap-strimzi-cluster-operator.yaml
```

Adding filters to the Topic Operator or User Operator

To add filters to the Topic Operator or User Operator, create or edit a logging ConfigMap.

In this procedure a logging ConfigMap is created with filters for the Topic Operator. The same approach is used for the User Operator.

Procedure

Create the ConfigMap.

You can create the ConfigMap as a YAML file or from a properties file.

In this example, the filter properties return logs only for the my-topic topic:

kind: ConfigMap
apiVersion: v1
metadata:
  name: logging-configmap
data:
  log4j2.properties:
    rootLogger.level="INFO"
    appender.console.filter.filter1.type=MarkerFilter
    appender.console.filter.filter1.onMatch=ACCEPT
    appender.console.filter.filter1.onMismatch=DENY
    appender.console.filter.filter1.marker=KafkaTopic(my-namespace/my-topic)

If you are using a properties file, specify the file at the command line:

oc create configmap logging-configmap --from-file=log4j2.properties

The properties file defines the logging configuration:

# Define the logger
rootLogger.level="INFO"
# Set the filters
appender.console.filter.filter1.type=MarkerFilter
appender.console.filter.filter1.onMatch=ACCEPT
appender.console.filter.filter1.onMismatch=DENY
appender.console.filter.filter1.marker=KafkaTopic(my-namespace/my-topic)
# ...

Define external logging in the spec of the resource, setting the logging.valueFrom.configMapKeyRef.name to the name of the ConfigMap and logging.valueFrom.configMapKeyRef.key to the key in this ConfigMap.
For the Topic Operator, logging is specified in the topicOperator configuration of the Kafka resource.
```
spec:
  # ...
  entityOperator:
    topicOperator:
      logging:
        type: external
        valueFrom:
          configMapKeyRef:
            name: logging-configmap
            key: log4j2.properties
```
Apply the changes by deploying the Cluster Operator:

create -f install/cluster-operator -n my-cluster-operator-namespace

Additional resources

Configuring Kafka
Cluster Operator logging
Topic Operator logging
User Operator logging

Chapter 3. Loading configuration values from external sources

Use configuration provider plugins to load configuration data from external sources. The providers operate independently of AMQ Streams. You can use them to load configuration data for all Kafka components, including producers and consumers. Use them, for example, to provide the credentials for Kafka Connect connector configuration.

OpenShift Configuration Provider

The OpenShift Configuration Provider plugin loads configuration data from OpenShift secrets or ConfigMaps.

Suppose you have a Secret object that’s managed outside the Kafka namespace, or outside the Kafka cluster. The OpenShift Configuration Provider allows you to reference the values of the secret in your configuration without extracting the files. You just need to tell the provider what secret to use and provide access rights. The provider loads the data without needing to restart the Kafka component, even when using a new Secret or ConfigMap object. This capability avoids disruption when a Kafka Connect instance hosts multiple connectors.

Environment Variables Configuration Provider

The Environment Variables Configuration Provider plugin loads configuration data from environment variables.

The values for the environment variables can be mapped from secrets or ConfigMaps. You can use the Environment Variables Configuration Provider, for example, to load certificates or JAAS configuration from environment variables mapped from OpenShift secrets.

Note

OpenShift Configuration Provider can’t use mounted files. For example, it can’t load values that need the location of a truststore or keystore. Instead, you can mount ConfigMaps or secrets into a Kafka Connect pod as environment variables or volumes. You can use the Environment Variables Configuration Provider to load values for environment variables. You add configuration using the externalConfiguration property in KafkaConnect.spec. You don’t need to set up access rights with this approach. However, Kafka Connect will need a restart when using a new Secret or ConfigMap for a connector. This will cause disruption to all the Kafka Connect instance’s connectors.

3.1. Loading configuration values from a ConfigMap

This procedure shows how to use the OpenShift Configuration Provider plugin.

In the procedure, an external ConfigMap object provides configuration properties for a connector.

Prerequisites

An OpenShift cluster is available.
A Kafka cluster is running.
The Cluster Operator is running.

Procedure

Create a ConfigMap or Secret that contains the configuration properties.
In this example, a ConfigMap object named my-connector-configuration contains connector properties:
Example ConfigMap with connector properties
```
apiVersion: v1
kind: ConfigMap
metadata:
  name: my-connector-configuration
data:
  option1: value1
  option2: value2
```
Specify the OpenShift Configuration Provider in the Kafka Connect configuration.
The specification shown here can support loading values from secrets and ConfigMaps.
Example Kafka Connect configuration to enable the OpenShift Configuration Provider
```
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  # ...
  config:
    # ...
    config.providers: secrets,configmaps 1
    config.providers.secrets.class: io.strimzi.kafka.KubernetesSecretConfigProvider 2
    config.providers.configmaps.class: io.strimzi.kafka.KubernetesConfigMapConfigProvider 3
  # ...
```
1
The alias for the configuration provider is used to define other configuration parameters. The provider parameters use the alias from config.providers, taking the form config.providers.${alias}.class.
2
KubernetesSecretConfigProvider provides values from secrets.
3
KubernetesConfigMapConfigProvider provides values from config maps.
Create or update the resource to enable the provider.
```
oc apply -f <kafka_connect_configuration_file>
```

Create a role that permits access to the values in the external config map.

Example role to access values from a config map

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: connector-configuration-role
rules:
- apiGroups: [""]
  resources: ["configmaps"]
  resourceNames: ["my-connector-configuration"]
  verbs: ["get"]
# ...

The rule gives the role permission to access the my-connector-configuration config map.

Create a role binding to permit access to the namespace that contains the config map.
Example role binding to access the namespace that contains the config map
```
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: connector-configuration-role-binding
subjects:
- kind: ServiceAccount
  name: my-connect-connect
  namespace: my-project
roleRef:
  kind: Role
  name: connector-configuration-role
  apiGroup: rbac.authorization.k8s.io
# ...
```
The role binding gives the role permission to access the my-project namespace.
The service account must be the same one used by the Kafka Connect deployment. The service account name format is <cluster_name>-connect, where <cluster_name> is the name of the KafkaConnect custom resource.
Reference the config map in the connector configuration.
Example connector configuration referencing the config map
```
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-connector
  labels:
    strimzi.io/cluster: my-connect
spec:
  # ...
  config:
    option: ${configmaps:my-project/my-connector-configuration:option1}
    # ...
# ...
```
Placeholders for the property values in the config map are referenced in the connector configuration. The placeholder structure is configmaps:<path_and_file_name>:<property>. KubernetesConfigMapConfigProvider reads and extracts the option1 property value from the external config map.

3.2. Loading configuration values from environment variables

This procedure shows how to use the Environment Variables Configuration Provider plugin.

In the procedure, environment variables provide configuration properties for a connector. A database password is specified as an environment variable.

Prerequisites

An OpenShift cluster is available.
A Kafka cluster is running.
The Cluster Operator is running.

Procedure

Specify the Environment Variables Configuration Provider in the Kafka Connect configuration.
Define environment variables using the externalConfiguration property.
Example Kafka Connect configuration to enable the Environment Variables Configuration Provider
```
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  # ...
  config:
    # ...
    config.providers: env 1
    config.providers.env.class: io.strimzi.kafka.EnvVarConfigProvider 2
  # ...
  externalConfiguration:
    env:
      - name: DB_PASSWORD 3
        valueFrom:
          secretKeyRef:
            name: db-creds 4
            key: dbPassword 5
  # ...
```
1
The alias for the configuration provider is used to define other configuration parameters. The provider parameters use the alias from config.providers, taking the form config.providers.${alias}.class.
2
EnvVarConfigProvider provides values from environment variables.
3
The DB_PASSWORD environment variable takes a password value from a secret.
4
The name of the secret containing the predefined password.
5
The key for the password stored inside the secret.
Create or update the resource to enable the provider.
```
oc apply -f <kafka_connect_configuration_file>
```

Reference the environment variable in the connector configuration.

Example connector configuration referencing the environment variable

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-connector
  labels:
    strimzi.io/cluster: my-connect
spec:
  # ...
  config:
    option: ${env:DB_PASSWORD}
    # ...
# ...

Chapter 4. Applying security context to AMQ Streams pods and containers

Security context defines constraints on pods and containers. By specifying a security context, pods and containers only have the permissions they need. For example, permissions can control runtime operations or access to resources.

4.1. Handling of security context by OpenShift platform

Handling of security context depends on the tooling of the OpenShift platform you are using.

For example, 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.

By default, OpenShift injects security context configuration automatically. In most cases, this means you don’t need to configure security context for the pods and containers created by the Cluster Operator. Although you can still create and manage your own SCCs.

For more information, see the OpenShift documentation.

Chapter 5. Accessing Kafka outside of the OpenShift cluster

Use an external listener to expose your AMQ Streams Kafka cluster to a client outside an OpenShift environment.

Specify the connection type to expose Kafka in the external listener configuration.

nodeport uses a NodePort type Service
loadbalancer uses a Loadbalancer type Service
ingress uses Kubernetes Ingress and the Ingress NGINX Controller for Kubernetes
route uses OpenShift Routes and the HAProxy router

For more information on listener configuration, see GenericKafkaListener schema reference.

If you want to know more about the pros and cons of each connection type, refer to Accessing Apache Kafka in Strimzi.

Note

route is only supported on OpenShift

5.1. Accessing Kafka using node ports

This procedure describes how to access an AMQ Streams Kafka cluster from an external client using node ports.

To connect to a broker, you need a hostname and port number for the Kafka bootstrap address, as well as the certificate used for authentication.

Prerequisites

An OpenShift cluster
A running Cluster Operator

Procedure

Configure a Kafka resource with an external listener set to the nodeport type.

For example:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    # ...
    listeners:
      - name: external
        port: 9094
        type: nodeport
        tls: true
        authentication:
          type: tls
        # ...
    # ...
  zookeeper:
    # ...

Create or update the resource.
```
oc apply -f <kafka_configuration_file>
```
NodePort type services are created for each Kafka broker, as well as an external bootstrap service. The bootstrap service routes external traffic to the Kafka brokers. Node addresses used for connection are propagated to the status of the Kafka custom resource.
The cluster CA certificate to verify the identity of the kafka brokers is also created in the secret <cluster_name>-cluster-ca-cert.

Retrieve the bootstrap address you can use to access the Kafka cluster from the status of the Kafka resource.

oc get kafka <kafka_cluster_name> -o=jsonpath='{.status.listeners[?(@.name=="<listener_name>")].bootstrapServers}{"\n"}'

For example:

oc get kafka my-cluster -o=jsonpath='{.status.listeners[?(@.name=="external")].bootstrapServers}{"\n"}'

If TLS encryption is enabled, extract the public certificate of the broker certification authority.
```
oc get secret KAFKA-CLUSTER-NAME-cluster-ca-cert -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt
```
Use the extracted certificate in your Kafka client to configure TLS connection. If you enabled any authentication, you will also need to configure it in your client.

5.2. Accessing Kafka using loadbalancers

This procedure describes how to access an AMQ Streams Kafka cluster from an external client using loadbalancers.

To connect to a broker, you need the address of the bootstrap loadbalancer, as well as the certificate used for TLS encryption.

Prerequisites

An OpenShift cluster
A running Cluster Operator

Procedure

Configure a Kafka resource with an external listener set to the loadbalancer type.

For example:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    # ...
    listeners:
      - name: external
        port: 9094
        type: loadbalancer
        tls: true
        # ...
    # ...
  zookeeper:
    # ...

Create or update the resource.
```
oc apply -f <kafka_configuration_file>
```
loadbalancer type services and loadbalancers are created for each Kafka broker, as well as an external bootstrap service. The bootstrap service routes external traffic to all Kafka brokers. DNS names and IP addresses used for connection are propagated to the status of each service.
The cluster CA certificate to verify the identity of the kafka brokers is also created in the secret <cluster_name>-cluster-ca-cert.

Retrieve the address of the bootstrap service you can use to access the Kafka cluster from the status of the Kafka resource.

oc get kafka <kafka_cluster_name> -o=jsonpath='{.status.listeners[?(@.name=="<listener_name>")].bootstrapServers}{"\n"}'

For example:

oc get kafka my-cluster -o=jsonpath='{.status.listeners[?(@.name=="external")].bootstrapServers}{"\n"}'

If TLS encryption is enabled, extract the public certificate of the broker certification authority.
```
oc get secret KAFKA-CLUSTER-NAME-cluster-ca-cert -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt
```
Use the extracted certificate in your Kafka client to configure the TLS connection. If you enabled any authentication, you will also need to configure it in your client.

5.3. Accessing Kafka using an Ingress NGINX Controller for OpenShift

Use an Ingress NGINX Controller for Kubernetes to access an AMQ Streams Kafka cluster from clients outside the OpenShift cluster.

To be able to use an Ingress NGINX Controller for OpenShift, add configuration for an ingress type listener in the Kafka custom resource. When applied, the configuration creates a dedicated ingress and service for an external bootstrap and each broker in the cluster. Clients connect to the bootstrap ingress, which routes them through the bootstrap service to connect to a broker. Per-broker connections are then established using DNS names, which route traffic from the client to the broker through the broker-specific ingresses and services.

To connect to a broker, you specify a hostname for the ingress bootstrap address, as well as the TLS certificate. Authentication is optional.

For access using an ingress, the port used in the Kafka client is typically 443.

TLS passthrough

Make sure that you enable TLS passthrough in your Ingress NGINX Controller for OpenShift deployment. Kafka uses a binary protocol over TCP, but the Ingress NGINX Controller for OpenShift is designed to work with a HTTP protocol. To be able to route TCP traffic through ingresses, AMQ Streams uses TLS passthrough with Server Name Indication (SNI).

SNI helps with identifying and passing connection to Kafka brokers. In passthrough mode, TLS encryption is always used. Because the connection passes to the brokers, the listeners use the TLS certificates signed by the internal cluster CA and not the ingress certificates. To configure listeners to use your own listener certificates, use the brokerCertChainAndKey property.

For more information about enabling TLS passthrough, see the TLS passthrough documentation.

Prerequisites

An Ingress NGINX Controller for OpenShift is running with TLS passthrough enabled
A running Cluster Operator

In this procedure, the Kafka cluster name is my-cluster.

Procedure

Configure a Kafka resource with an external listener set to the ingress type.

Specify an ingress hostname for the bootstrap service and each of the Kafka brokers in the Kafka cluster. Add any hostname to the bootstrap and broker-<index> prefixes that identify the bootstrap and brokers.

For example:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  labels:
    app: my-cluster
  name: my-cluster
  namespace: myproject
spec:
  kafka:
    # ...
    listeners:
      - name: external
        port: 9094
        type: ingress
        tls: true 1
        authentication:
          type: tls
        configuration:
          bootstrap:
            host: bootstrap.myingress.com
          brokers:
          - broker: 0
            host: broker-0.myingress.com
          - broker: 1
            host: broker-1.myingress.com
          - broker: 2
            host: broker-2.myingress.com
          class: nginx  2
    # ...
  zookeeper:
    # ...

1: For ingress type listeners, TLS encryption must be enabled (true).
2: (Optional) Class that specifies the ingress controller to use. You might need to add a class if you have not set up a default and a class name is missing in the ingresses created.

Create or update the resource.

oc apply -f <kafka_configuration_file>

A cluster CA certificate to verify the identity of the kafka brokers is created in the secret my-cluster-cluster-ca-cert.

ClusterIP type services are created for each Kafka broker, as well as an external bootstrap service.

An ingress is also created for each service, with a DNS address to expose them using the Ingress NGINX Controller for OpenShift.

Ingresses created for the bootstrap and brokers

NAME                        CLASS  HOSTS                    ADDRESS               PORTS
my-cluster-kafka-0          nginx  broker-0.myingress.com   external.ingress.com  80,443
my-cluster-kafka-1          nginx  broker-1.myingress.com   external.ingress.com  80,443
my-cluster-kafka-2          nginx  broker-2.myingress.com   external.ingress.com  80,443
my-cluster-kafka-bootstrap  nginx  bootstrap.myingress.com  external.ingress.com  80,443

The DNS addresses used for client connection are propagated to the status of each ingress.

Status for the bootstrap ingress

status:
  loadBalancer:
    ingress:
      - hostname: external.ingress.com
 # ...

Use a target broker to check the client-server TLS connection on port 443 using the OpenSSL s_client.
```
openssl s_client -connect broker-0.myingress.com:443 -servername broker-0.myingress.com -showcerts
```
The server name is the SNI for passing the connection to the broker.
If the connection is successful, the certificates for the broker are returned.
Certificates for the broker
```
Certificate chain
 0 s:O = io.strimzi, CN = my-cluster-kafka
   i:O = io.strimzi, CN = cluster-ca v0
```

Extract the cluster CA certificate.

oc get secret my-cluster-cluster-ca-cert -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt

Configure your client to connect to the brokers.
1. Specify the bootstrap host (from the listener configuration) and port 443 in your Kafka client as the bootstrap address to connect to the Kafka cluster. For example, bootstrap.myingress.com:443.
2. Add the extracted certificate to the truststore of your Kafka client to configure a TLS connection.
  If you enabled any authentication, you will also need to configure it in your client.

Note

If you are using your own listener certificates, check whether you need to add the CA certificate to the client’s truststore configuration. If it is a public (external) CA, you usually won’t need to add it.

5.4. Accessing Kafka using OpenShift routes

Use OpenShift routes to access an AMQ Streams Kafka cluster from clients outside the OpenShift cluster.

To be able to use routes, add configuration for a route type listener in the Kafka custom resource. When applied, the configuration creates a dedicated route and service for an external bootstrap and each broker in the cluster. Clients connect to the bootstrap route, which routes them through the bootstrap service to connect to a broker. Per-broker connections are then established using DNS names, which route traffic from the client to the broker through the broker-specific routes and services.

To connect to a broker, you specify a hostname for the route bootstrap address, as well as the certificate used for authentication.

For access using routes, the port is always 443.

Warning

An OpenShift route address comprises the name of the Kafka cluster, the name of the listener, and the name of the project it is created in. For example, my-cluster-kafka-listener1-bootstrap-myproject (<cluster_name>-kafka-<listener_name>-bootstrap-<namespace>). Be careful that the whole length of the address does not exceed a maximum limit of 63 characters.

TLS passthrough

TLS passthrough is enabled for routes created by AMQ Streams. Kafka uses a binary protocol over TCP, but routes are designed to work with a HTTP protocol. To be able to route TCP traffic through routes, AMQ Streams uses TLS passthrough with Server Name Indication (SNI).

SNI helps with identifying and passing connection to Kafka brokers. In passthrough mode, TLS encryption is always used. Because the connection passes to the brokers, the listeners use TLS certificates signed by the internal cluster CA and not the ingress certificates. To configure listeners to use your own listener certificates, use the brokerCertChainAndKey property.

Prerequisites

A running Cluster Operator

In this procedure, the Kafka cluster name is my-cluster.

Procedure

Configure a Kafka resource with an external listener set to the route type.

For example:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  labels:
    app: my-cluster
  name: my-cluster
  namespace: myproject
spec:
  kafka:
    # ...
    listeners:
      - name: listener1
        port: 9094
        type: route
        tls: true 1
        # ...
    # ...
  zookeeper:
    # ...

1: For route type listeners, TLS encryption must be enabled (true).

Create or update the resource.

oc apply -f <kafka_configuration_file>

A cluster CA certificate to verify the identity of the kafka brokers is created in the secret my-cluster-cluster-ca-cert.

ClusterIP type services are created for each Kafka broker, as well as an external bootstrap service.

A route is also created for each service, with a DNS address (host/port) to expose them using the default OpenShift HAProxy router.

The routes are preconfigured with TLS passthrough.

Routes created for the bootstraps and brokers

NAME                                  HOST/PORT                                                   SERVICES                              PORT  TERMINATION
my-cluster-kafka-listener1-0          my-cluster-kafka-listener1-0-my-project.router.com          my-cluster-kafka-listener1-0          9094  passthrough
my-cluster-kafka-listener1-1          my-cluster-kafka-listener1-1-my-project.router.com          my-cluster-kafka-listener1-1          9094  passthrough
my-cluster-kafka-listener1-2          my-cluster-kafka-listener1-2-my-project.router.com          my-cluster-kafka-listener1-2          9094  passthrough
my-cluster-kafka-listener1-bootstrap  my-cluster-kafka-listener1-bootstrap-my-project.router.com  my-cluster-kafka-listener1-bootstrap  9094  passthrough

The DNS addresses used for client connection are propagated to the status of each route.

Example status for the bootstrap route

status:
  ingress:
    - host: >-
        my-cluster-kafka-listener1-bootstrap-my-project.router.com
 # ...

Use a target broker to check the client-server TLS connection on port 443 using the OpenSSL s_client.
```
openssl s_client -connect my-cluster-kafka-listener1-0-my-project.router.com:443 -servername my-cluster-kafka-listener1-0-my-project.router.com -showcerts
```
The server name is the SNI for passing the connection to the broker.
If the connection is successful, the certificates for the broker are returned.
Certificates for the broker
```
Certificate chain
 0 s:O = io.strimzi, CN = my-cluster-kafka
   i:O = io.strimzi, CN = cluster-ca v0
```
Retrieve the address of the bootstrap service from the status of the Kafka resource.
```
oc get kafka my-cluster -o=jsonpath='{.status.listeners[?(@.name=="listener1")].bootstrapServers}{"\n"}'

my-cluster-kafka-listener1-bootstrap-my-project.router.com:443
```
The address comprises the cluster name, the listener name, the project name and the domain of the router (router.com in this example).

Extract the cluster CA certificate.

oc get secret my-cluster-cluster-ca-cert -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt

Configure your client to connect to the brokers.
1. Specify the address for the bootstrap service and port 443 in your Kafka client as the bootstrap address to connect to the Kafka cluster.
2. Add the extracted certificate to the truststore of your Kafka client to configure a TLS connection.
  If you enabled any authentication, you will also need to configure it in your client.

Note

Chapter 6. Managing secure access to Kafka

You can secure your Kafka cluster by managing the access each client has to the Kafka brokers.

A secure connection between Kafka brokers and clients can encompass:

Encryption for data exchange
Authentication to prove identity
Authorization to allow or decline actions executed by users

This chapter explains how to set up secure connections between Kafka brokers and clients, with sections describing:

Security options for Kafka clusters and clients
How to secure Kafka brokers
How to use an authorization server for OAuth 2.0 token-based authentication and authorization

6.1. Security options for Kafka

Use the Kafka resource to configure the mechanisms used for Kafka authentication and authorization.

6.1.1. Listener authentication

Configure client authentication for Kafka brokers by creating listeners. Specify the listener authentication type using the Kafka.spec.kafka.listeners.authentication property in the Kafka resource.

For clients inside the OpenShift cluster, you can create plain (without encryption) or tls internal listeners. The internal listener type use a headless service and the DNS names given to the broker pods. As an alternative to the headless service, you can also create a cluster-ip type of internal listener to expose Kafka using per-broker ClusterIP services. For clients outside the OpenShift cluster, you create external listeners and specify a connection mechanism, which can be nodeport, loadbalancer, ingress, or route (on OpenShift).

For more information on the configuration options for connecting an external client, see Accessing Kafka outside of the OpenShift cluster.

Supported authentication options:

mTLS authentication (only on the listeners with TLS enabled encryption)
SCRAM-SHA-512 authentication
OAuth 2.0 token-based authentication
Custom authentication

The authentication option you choose depends on how you wish to authenticate client access to Kafka brokers.

Note

Try exploring the standard authentication options before using custom authentication. Custom authentication allows for any type of kafka-supported authentication. It can provide more flexibility, but also adds complexity.

Figure 6.1. Kafka listener authentication options

options for listener authentication configuration

The listener authentication property is used to specify an authentication mechanism specific to that listener.

If no authentication property is specified then the listener does not authenticate clients which connect through that listener. The listener will accept all connections without authentication.

Authentication must be configured when using the User Operator to manage KafkaUsers.

The following example shows:

A plain listener configured for SCRAM-SHA-512 authentication
A tls listener with mTLS authentication
An external listener with mTLS authentication

Each listener is configured with a unique name and port within a Kafka cluster.

Note

Listeners cannot be configured to use the ports reserved for inter-broker communication (9091 or 9090) and metrics (9404).

Example listener authentication configuration

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  namespace: myproject
spec:
  kafka:
    # ...
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: true
        authentication:
          type: scram-sha-512
      - name: tls
        port: 9093
       type: internal
       tls: true
       authentication:
         type: tls
      - name: external
        port: 9094
        type: loadbalancer
        tls: true
        authentication:
          type: tls
# ...

6.1.1.1. mTLS authentication

mTLS authentication is always used for the communication between Kafka brokers and ZooKeeper pods.

AMQ Streams can configure Kafka to use TLS (Transport Layer Security) to provide encrypted communication between Kafka brokers and clients either with or without mutual authentication. For mutual, or two-way, authentication, both the server and the client present certificates. When you configure mTLS authentication, the broker authenticates the client (client authentication) and the client authenticates the broker (server authentication).

mTLS listener configuration in the Kafka resource requires the following:

tls: true to specify TLS encryption and server authentication
authentication.type: tls to specify the client authentication

When a Kafka cluster is created by the Cluster Operator, it creates a new secret with the name <cluster_name>-cluster-ca-cert. The secret contains a CA certificate. The CA certificate is in PEM and PKCS #12 format. To verify a Kafka cluster, add the CA certificate to the truststore in your client configuration. To verify a client, add a user certificate and key to the keystore in your client configuration. For more information on configuring a client for mTLS, see Section 6.2.2, “User authentication”.

Note

TLS authentication is more commonly one-way, with one party authenticating the identity of another. For example, when HTTPS is used between a web browser and a web server, the browser obtains proof of the identity of the web server.

6.1.1.2. SCRAM-SHA-512 authentication

SCRAM (Salted Challenge Response Authentication Mechanism) is an authentication protocol that can establish mutual authentication using passwords. AMQ Streams can configure Kafka to use SASL (Simple Authentication and Security Layer) SCRAM-SHA-512 to provide authentication on both unencrypted and encrypted client connections.

When SCRAM-SHA-512 authentication is used with a TLS connection, the TLS protocol provides the encryption, but is not used for authentication.

The following properties of SCRAM make it safe to use SCRAM-SHA-512 even on unencrypted connections:

The passwords are not sent in the clear over the communication channel. Instead the client and the server are each challenged by the other to offer proof that they know the password of the authenticating user.
The server and client each generate a new challenge for each authentication exchange. This means that the exchange is resilient against replay attacks.

When KafkaUser.spec.authentication.type is configured with scram-sha-512 the User Operator will generate a random 12-character password consisting of upper and lowercase ASCII letters and numbers.

6.1.1.3. Network policies

By default, AMQ Streams automatically creates a NetworkPolicy resource for every listener that is enabled on a Kafka broker. This NetworkPolicy allows applications to connect to listeners in all namespaces. Use network policies as part of the listener configuration.

If you want to restrict access to a listener at the network level to only selected applications or namespaces, use the networkPolicyPeers property. Each listener can have a different networkPolicyPeers configuration. For more information on network policy peers, refer to the NetworkPolicyPeer API reference.

If you want to use custom network policies, you can set the STRIMZI_NETWORK_POLICY_GENERATION environment variable to false in the Cluster Operator configuration. For more information, see Cluster Operator configuration.

Note

Your configuration of OpenShift must support ingress NetworkPolicies in order to use network policies in AMQ Streams.

6.1.1.4. Additional listener configuration options

You can use the properties of the GenericKafkaListenerConfiguration schema to add further configuration to listeners.

6.1.2. Kafka authorization

Configure authorization for Kafka brokers using the Kafka.spec.kafka.authorization property in the Kafka resource. If the authorization property is missing, no authorization is enabled and clients have no restrictions. When enabled, authorization is applied to all enabled listeners. The authorization method is defined in the type field.

Supported authorization options:

Simple authorization
OAuth 2.0 authorization (if you are using OAuth 2.0 token based authentication)
Open Policy Agent (OPA) authorization
Custom authorization

Figure 6.2. Kafka cluster authorization options

options for kafka authorization configuration

6.1.2.1. Super users

Super users can access all resources in your Kafka cluster regardless of any access restrictions, and are supported by all authorization mechanisms.

To designate super users for a Kafka cluster, add a list of user principals to the superUsers property. If a user uses mTLS authentication, the username is the common name from the TLS certificate subject prefixed with CN=. If you are not using the User Operator and using your own certificates for mTLS, the username is the full certificate subject. A full certificate subject can have the following fields: CN=user,OU=my_ou,O=my_org,L=my_location,ST=my_state,C=my_country_code. Omit any fields that are not present.

An example configuration with super users

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  namespace: myproject
spec:
  kafka:
    # ...
    authorization:
      type: simple
      superUsers:
        - CN=client_1
        - user_2
        - CN=client_3
        - CN=client_4,OU=my_ou,O=my_org,L=my_location,ST=my_state,C=US
        - CN=client_5,OU=my_ou,O=my_org,C=GB
        - CN=client_6,O=my_org
    # ...

6.2. Security options for Kafka clients

Use the KafkaUser resource to configure the authentication mechanism, authorization mechanism, and access rights for Kafka clients. In terms of configuring security, clients are represented as users.

You can authenticate and authorize user access to Kafka brokers. Authentication permits access, and authorization constrains the access to permissible actions.

You can also create super users that have unconstrained access to Kafka brokers.

The authentication and authorization mechanisms must match the specification for the listener used to access the Kafka brokers.

Configuring users for secure access to Kafka brokers

For more information on configuring a KafkaUser resource to access Kafka brokers securely, see the following sections:

6.2.1. Identifying a Kafka cluster for user handling

A KafkaUser resource includes a label that defines the appropriate name of the Kafka cluster (derived from the name of the Kafka resource) to which it belongs.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster

The label is used by the User Operator to identify the KafkaUser resource and create a new user, and also in subsequent handling of the user.

If the label does not match the Kafka cluster, the User Operator cannot identify the KafkaUser and the user is not created.

If the status of the KafkaUser resource remains empty, check your label.

6.2.2. User authentication

Use the KafkaUser custom resource to configure authentication credentials for users (clients) that require access to a Kafka cluster. Configure the credentials using the authentication property in KafkaUser.spec. By specifying a type, you control what credentials are generated.

Supported authentication types:

tls for mTLS authentication
tls-external for mTLS authentication using external certificates
scram-sha-512 for SCRAM-SHA-512 authentication

If tls or scram-sha-512 is specified, the User Operator creates authentication credentials when it creates the user. If tls-external is specified, the user still uses mTLS, but no authentication credentials are created. Use this option when you’re providing your own certificates. When no authentication type is specified, the User Operator does not create the user or its credentials.

You can use tls-external to authenticate with mTLS using a certificate issued outside the User Operator. The User Operator does not generate a TLS certificate or a secret. You can still manage ACL rules and quotas through the User Operator in the same way as when you’re using the tls mechanism. This means that you use the CN=USER-NAME format when specifying ACL rules and quotas. USER-NAME is the common name given in a TLS certificate.

6.2.2.1. mTLS authentication

To use mTLS authentication, you set the type field in the KafkaUser resource to tls.

Example user with mTLS authentication enabled

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: tls
  # ...

The authentication type must match the equivalent configuration for the Kafka listener used to access the Kafka cluster.

When the user is created by the User Operator, it creates a new secret with the same name as the KafkaUser resource. The secret contains a private and public key for mTLS. The public key is contained in a user certificate, which is signed by a clients CA (certificate authority) when it is created. All keys are in X.509 format.

Note

If you are using the clients CA generated by the Cluster Operator, the user certificates generated by the User Operator are also renewed when the clients CA is renewed by the Cluster Operator.

The user secret provides keys and certificates in PEM and PKCS #12 formats.

Example secret with user credentials

apiVersion: v1
kind: Secret
metadata:
  name: my-user
  labels:
    strimzi.io/kind: KafkaUser
    strimzi.io/cluster: my-cluster
type: Opaque
data:
  ca.crt: <public_key> # Public key of the clients CA
  user.crt: <user_certificate> # Public key of the user
  user.key: <user_private_key> # Private key of the user
  user.p12: <store> # PKCS #12 store for user certificates and keys
  user.password: <password_for_store> # Protects the PKCS #12 store

When you configure a client, you specify the following:

Truststore properties for the public cluster CA certificate to verify the identity of the Kafka cluster
Keystore properties for the user authentication credentials to verify the client

The configuration depends on the file format (PEM or PKCS #12). This example uses PKCS #12 stores, and the passwords required to access the credentials in the stores.

Example client configuration using mTLS in PKCS #12 format

bootstrap.servers=<kafka_cluster_name>-kafka-bootstrap:9093 1
security.protocol=SSL 2
ssl.truststore.location=/tmp/ca.p12 3
ssl.truststore.password=<truststore_password> 4
ssl.keystore.location=/tmp/user.p12 5
ssl.keystore.password=<keystore_password> 6

1: The bootstrap server address to connect to the Kafka cluster.
2: The security protocol option when using TLS for encryption.
3: The truststore location contains the public key certificate (ca.p12) for the Kafka cluster. A cluster CA certificate and password is generated by the Cluster Operator in the <cluster_name>-cluster-ca-cert secret when the Kafka cluster is created.
4: The password (ca.password) for accessing the truststore.
5: The keystore location contains the public key certificate (user.p12) for the Kafka user.
6: The password (user.password) for accessing the keystore.

6.2.2.2. mTLS authentication using a certificate issued outside the User Operator

To use mTLS authentication using a certificate issued outside the User Operator, you set the type field in the KafkaUser resource to tls-external. A secret and credentials are not created for the user.

Example user with mTLS authentication that uses a certificate issued outside the User Operator

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: tls-external
  # ...

6.2.2.3. SCRAM-SHA-512 authentication

To use the SCRAM-SHA-512 authentication mechanism, you set the type field in the KafkaUser resource to scram-sha-512.

Example user with SCRAM-SHA-512 authentication enabled

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
  # ...

When the user is created by the User Operator, it creates a new secret with the same name as the KafkaUser resource. The secret contains the generated password in the password key, which is encoded with base64. In order to use the password, it must be decoded.

Example secret with user credentials

apiVersion: v1
kind: Secret
metadata:
  name: my-user
  labels:
    strimzi.io/kind: KafkaUser
    strimzi.io/cluster: my-cluster
type: Opaque
data:
  password: Z2VuZXJhdGVkcGFzc3dvcmQ= 1
  sasl.jaas.config: b3JnLmFwYWNoZS5rYWZrYS5jb21tb24uc2VjdXJpdHkuc2NyYW0uU2NyYW1Mb2dpbk1vZHVsZSByZXF1aXJlZCB1c2VybmFtZT0ibXktdXNlciIgcGFzc3dvcmQ9ImdlbmVyYXRlZHBhc3N3b3JkIjsK 2

1: The generated password, base64 encoded.
2: The JAAS configuration string for SASL SCRAM-SHA-512 authentication, base64 encoded.

Decoding the generated password:

echo "Z2VuZXJhdGVkcGFzc3dvcmQ=" | base64 --decode

6.2.2.3.1. Custom password configuration

When a user is created, AMQ Streams generates a random password. You can use your own password instead of the one generated by AMQ Streams. To do so, create a secret with the password and reference it in the KafkaUser resource.

Example user with a password set for SCRAM-SHA-512 authentication

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
    password:
      valueFrom:
        secretKeyRef:
          name: my-secret 1
          key: my-password 2
  # ...

1: The name of the secret containing the predefined password.
2: The key for the password stored inside the secret.

6.2.3. User authorization

Use the KafkaUser custom resource to configure authorization rules for users (clients) that require access to a Kafka cluster. Configure the rules using the authorization property in KafkaUser.spec. By specifying a type, you control what rules are used.

To use simple authorization, you set the type property to simple in KafkaUser.spec.authorization. The simple authorization uses the Kafka Admin API to manage the ACL rules inside your Kafka cluster. Whether ACL management in the User Operator is enabled or not depends on your authorization configuration in the Kafka cluster.

For simple authorization, ACL management is always enabled.
For OPA authorization, ACL management is always disabled. Authorization rules are configured in the OPA server.
For Red Hat Single Sign-On authorization, you can manage the ACL rules directly in Red Hat Single Sign-On. You can also delegate authorization to the simple authorizer as a fallback option in the configuration. When delegation to the simple authorizer is enabled, the User Operator will enable management of ACL rules as well.
For custom authorization using a custom authorization plugin, use the supportsAdminApi property in the .spec.kafka.authorization configuration of the Kafka custom resource to enable or disable the support.

Authorization is cluster-wide. The authorization type must match the equivalent configuration in the Kafka custom resource.

If ACL management is not enabled, AMQ Streams rejects a resource if it contains any ACL rules.

If you’re using a standalone deployment of the User Operator, ACL management is enabled by default. You can disable it using the STRIMZI_ACLS_ADMIN_API_SUPPORTED environment variable.

If no authorization is specified, the User Operator does not provision any access rights for the user. Whether such a KafkaUser can still access resources depends on the authorizer being used. For example, for the AclAuthorizer this is determined by its allow.everyone.if.no.acl.found configuration.

6.2.3.1. ACL rules

AclAuthorizer uses ACL rules to manage access to Kafka brokers.

ACL rules grant access rights to the user, which you specify in the acls property.

For more information about the AclRule object, see the AclRule schema reference.

6.2.3.2. Super user access to Kafka brokers

If a user is added to a list of super users in a Kafka broker configuration, the user is allowed unlimited access to the cluster regardless of any authorization constraints defined in ACLs in KafkaUser.

For more information on configuring super user access to brokers, see Kafka authorization.

6.2.3.3. User quotas

You can configure the spec for the KafkaUser resource to enforce quotas so that a user does not exceed a configured level of access to Kafka brokers. You can set size-based network usage and time-based CPU utilization thresholds. You can also add a partition mutation quota to control the rate at which requests to change partitions are accepted for user requests.

An example KafkaUser with user quotas

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  # ...
  quotas:
    producerByteRate: 1048576 1
    consumerByteRate: 2097152 2
    requestPercentage: 55 3
    controllerMutationRate: 10 4

1: Byte-per-second quota on the amount of data the user can push to a Kafka broker
2: Byte-per-second quota on the amount of data the user can fetch from a Kafka broker
3: CPU utilization limit as a percentage of time for a client group
4: Number of concurrent partition creation and deletion operations (mutations) allowed per second

For more information on these properties, see the KafkaUserQuotas schema reference.

6.3. Securing access to Kafka brokers

To establish secure access to Kafka brokers, you configure and apply:

A Kafka resource to:
- Create listeners with a specified authentication type
- Configure authorization for the whole Kafka cluster
A KafkaUser resource to access the Kafka brokers securely through the listeners

Configure the Kafka resource to set up:

Listener authentication
Network policies that restrict access to Kafka listeners
Kafka authorization
Super users for unconstrained access to brokers

Authentication is configured independently for each listener. Authorization is always configured for the whole Kafka cluster.

The Cluster Operator creates the listeners and sets up the cluster and client certificate authority (CA) certificates to enable authentication within the Kafka cluster.

You can replace the certificates generated by the Cluster Operator by installing your own certificates. You can also configure your listener to use a Kafka listener certificate managed by an external CA (certificate authority). Certificates are available in PKCS #12 format (.p12) and PEM (.crt) formats.

Use KafkaUser to enable the authentication and authorization mechanisms that a specific client uses to access Kafka.

Configure the KafkaUser resource to set up:

Authentication to match the enabled listener authentication
Authorization to match the enabled Kafka authorization
Quotas to control the use of resources by clients

The User Operator creates the user representing the client and the security credentials used for client authentication, based on the chosen authentication type.

Refer to the schema reference for more information on access configuration properties:

Kafka schema reference
KafkaUser schema reference
GenericKafkaListener schema reference

6.3.1. Securing Kafka brokers

This procedure shows the steps involved in securing Kafka brokers when running AMQ Streams.

The security implemented for Kafka brokers must be compatible with the security implemented for the clients requiring access.

Kafka.spec.kafka.listeners[*].authentication matches KafkaUser.spec.authentication
Kafka.spec.kafka.authorization matches KafkaUser.spec.authorization

The steps show the configuration for simple authorization and a listener using mTLS authentication. For more information on listener configuration, see GenericKafkaListener schema reference.

Alternatively, you can use SCRAM-SHA or OAuth 2.0 for listener authentication, and OAuth 2.0 or OPA for Kafka authorization.

Procedure

Configure the Kafka resource.
1. Configure the authorization property for authorization.
2. Configure the listeners property to create a listener with authentication.
  For example:
```
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    # ...
    authorization: 1
      type: simple
      superUsers: 2
        - CN=client_1
        - user_2
        - CN=client_3
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls 3
    # ...
  zookeeper:
    # ...
```
  1
  Authorization enables simple authorization on the Kafka broker using the AclAuthorizer Kafka plugin.
  2
  List of user principals with unlimited access to Kafka. CN is the common name from the client certificate when mTLS authentication is used.
  3
  Listener authentication mechanisms may be configured for each listener, and specified as mTLS, SCRAM-SHA-512, or token-based OAuth 2.0.
  If you are configuring an external listener, the configuration is dependent on the chosen connection mechanism.
Create or update the Kafka resource.
```
oc apply -f <kafka_configuration_file>
```
The Kafka cluster is configured with a Kafka broker listener using mTLS authentication.
A service is created for each Kafka broker pod.
A service is created to serve as the bootstrap address for connection to the Kafka cluster.
The cluster CA certificate to verify the identity of the kafka brokers is also created in the secret <cluster_name>-cluster-ca-cert.

6.3.2. Securing user access to Kafka

Create or modify a KafkaUser to represent a client that requires secure access to the Kafka cluster.

When you configure the KafkaUser authentication and authorization mechanisms, ensure they match the equivalent Kafka configuration:

KafkaUser.spec.authentication matches Kafka.spec.kafka.listeners[*].authentication
KafkaUser.spec.authorization matches Kafka.spec.kafka.authorization

This procedure shows how a user is created with mTLS authentication. You can also create a user with SCRAM-SHA authentication.

The authentication required depends on the type of authentication configured for the Kafka broker listener.

Note

Authentication between Kafka users and Kafka brokers depends on the authentication settings for each. For example, it is not possible to authenticate a user with mTLS if it is not also enabled in the Kafka configuration.

Prerequisites

A running Kafka cluster configured with a Kafka broker listener using mTLS authentication and TLS encryption.
A running User Operator (typically deployed with the Entity Operator).

The authentication type in KafkaUser should match the authentication configured in Kafka brokers.

Procedure

Configure the KafkaUser resource.

For example:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication: 1
    type: tls
  authorization:
    type: simple 2
    acls:
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operations:
          - Describe
          - Read
      - resource:
          type: group
          name: my-group
          patternType: literal
        operations:
          - Read

1: User authentication mechanism, defined as mutual tls or scram-sha-512.
2: Simple authorization, which requires an accompanying list of ACL rules.

Create or update the KafkaUser resource.
```
oc apply -f <user_config_file>
```
The user is created, as well as a Secret with the same name as the KafkaUser resource. The Secret contains a private and public key for mTLS authentication.

For information on configuring a Kafka client with properties for secure connection to Kafka brokers, see Setting up client access to a Kafka cluster using listeners.

6.3.3. Restricting access to Kafka listeners using network policies

You can restrict access to a listener to only selected applications by using the networkPolicyPeers property.

Prerequisites

An OpenShift cluster with support for Ingress NetworkPolicies.
The Cluster Operator is running.

Procedure

Open the Kafka resource.

In the networkPolicyPeers property, define the application pods or namespaces that will be allowed to access the Kafka cluster.

For example, to configure a tls listener to allow connections only from application pods with the label app set to kafka-client:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    # ...
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
        networkPolicyPeers:
          - podSelector:
              matchLabels:
                app: kafka-client
    # ...
  zookeeper:
    # ...

Create or update the resource.
Use oc apply:
```
oc apply -f your-file
```

Additional resources

6.4. Using OAuth 2.0 token-based authentication

AMQ Streams supports the use of OAuth 2.0 authentication using the OAUTHBEARER and PLAIN mechanisms.

OAuth 2.0 enables standardized token-based authentication and authorization between applications, using a central authorization server to issue tokens that grant limited access to resources.

You can configure OAuth 2.0 authentication, then OAuth 2.0 authorization.

Kafka brokers and clients both need to be configured to use OAuth 2.0. OAuth 2.0 authentication can also be used in conjunction with simple or OPA-based Kafka authorization.

Using OAuth 2.0 token-based authentication, application clients can access resources on application servers (called resource servers) without exposing account credentials.

The application client passes an access token as a means of authenticating, which application servers can also use to determine the level of access to grant. The authorization server handles the granting of access and inquiries about access.

In the context of AMQ Streams:

Kafka brokers act as OAuth 2.0 resource servers
Kafka clients act as OAuth 2.0 application clients

Kafka clients authenticate to Kafka brokers. The brokers and clients communicate with the OAuth 2.0 authorization server, as necessary, to obtain or validate access tokens.

For a deployment of AMQ Streams, OAuth 2.0 integration provides:

Server-side OAuth 2.0 support for Kafka brokers
Client-side OAuth 2.0 support for Kafka MirrorMaker, Kafka Connect and the Kafka Bridge

6.4.1. OAuth 2.0 authentication mechanisms

AMQ Streams supports the OAUTHBEARER and PLAIN mechanisms for OAuth 2.0 authentication. Both mechanisms allow Kafka clients to establish authenticated sessions with Kafka brokers. The authentication flow between clients, the authorization server, and Kafka brokers is different for each mechanism.

We recommend that you configure clients to use OAUTHBEARER whenever possible. OAUTHBEARER provides a higher level of security than PLAIN because client credentials are never shared with Kafka brokers. Consider using PLAIN only with Kafka clients that do not support OAUTHBEARER.

You configure Kafka broker listeners to use OAuth 2.0 authentication for connecting clients. If necessary, you can use the OAUTHBEARER and PLAIN mechanisms on the same oauth listener. The properties to support each mechanism must be explicitly specified in the oauth listener configuration.

OAUTHBEARER overview

OAUTHBEARER is automatically enabled in the oauth listener configuration for the Kafka broker. You can set the enableOauthBearer property to true, though this is not required.

  # ...
  authentication:
    type: oauth
    # ...
    enableOauthBearer: true

Many Kafka client tools use libraries that provide basic support for OAUTHBEARER at the protocol level. To support application development, AMQ Streams provides an OAuth callback handler for the upstream Kafka Client Java libraries (but not for other libraries). Therefore, you do not need to write your own callback handlers. An application client can use the callback handler to provide the access token. Clients written in other languages, such as Go, must use custom code to connect to the authorization server and obtain the access token.

With OAUTHBEARER, the client initiates a session with the Kafka broker for credentials exchange, where credentials take the form of a bearer token provided by the callback handler. Using the callbacks, you can configure token provision in one of three ways:

Client ID and Secret (by using the OAuth 2.0 client credentials mechanism)
A long-lived access token, obtained manually at configuration time
A long-lived refresh token, obtained manually at configuration time

Note

OAUTHBEARER authentication can only be used by Kafka clients that support the OAUTHBEARER mechanism at the protocol level.

PLAIN overview

To use PLAIN, you must enable it in the oauth listener configuration for the Kafka broker.

In the following example, PLAIN is enabled in addition to OAUTHBEARER, which is enabled by default. If you want to use PLAIN only, you can disable OAUTHBEARER by setting enableOauthBearer to false.

  # ...
  authentication:
    type: oauth
    # ...
    enablePlain: true
    tokenEndpointUri: https://OAUTH-SERVER-ADDRESS/auth/realms/external/protocol/openid-connect/token

PLAIN is a simple authentication mechanism used by all Kafka client tools. To enable PLAIN to be used with OAuth 2.0 authentication, AMQ Streams provides OAuth 2.0 over PLAIN server-side callbacks.

With the AMQ Streams implementation of PLAIN, the client credentials are not stored in ZooKeeper. Instead, client credentials are handled centrally behind a compliant authorization server, similar to when OAUTHBEARER authentication is used.

When used with the OAuth 2.0 over PLAIN callbacks, Kafka clients authenticate with Kafka brokers using either of the following methods:

Client ID and secret (by using the OAuth 2.0 client credentials mechanism)
A long-lived access token, obtained manually at configuration time

For both methods, the client must provide the PLAIN username and password properties to pass credentials to the Kafka broker. The client uses these properties to pass a client ID and secret or username and access token.

Client IDs and secrets are used to obtain access tokens.

Access tokens are passed as password property values. You pass the access token with or without an $accessToken: prefix.

If you configure a token endpoint (tokenEndpointUri) in the listener configuration, you need the prefix.
If you don’t configure a token endpoint (tokenEndpointUri) in the listener configuration, you don’t need the prefix. The Kafka broker interprets the password as a raw access token.

If the password is set as the access token, the username must be set to the same principal name that the Kafka broker obtains from the access token. You can specify username extraction options in your listener using the userNameClaim, fallbackUserNameClaim, fallbackUsernamePrefix, and userInfoEndpointUri properties. The username extraction process also depends on your authorization server; in particular, how it maps client IDs to account names.

Note

OAuth over PLAIN does not support password grant mechanism. You can only 'proxy' through SASL PLAIN mechanism the client credentials (clientId + secret) or the access token as described above.

Additional resources

Section 6.4.6.2, “Configuring OAuth 2.0 support for Kafka brokers”

6.4.2. OAuth 2.0 Kafka broker configuration

Kafka broker configuration for OAuth 2.0 involves:

Creating the OAuth 2.0 client in the authorization server
Configuring OAuth 2.0 authentication in the Kafka custom resource

Note

In relation to the authorization server, Kafka brokers and Kafka clients are both regarded as OAuth 2.0 clients.

6.4.2.1. OAuth 2.0 client configuration on an authorization server

To configure a Kafka broker to validate the token received during session initiation, the recommended approach is to create an OAuth 2.0 client definition in an authorization server, configured as confidential, with the following client credentials enabled:

Client ID of kafka (for example)
Client ID and Secret as the authentication mechanism

Note

You only need to use a client ID and secret when using a non-public introspection endpoint of the authorization server. The credentials are not typically required when using public authorization server endpoints, as with fast local JWT token validation.

6.4.2.2. OAuth 2.0 authentication configuration in the Kafka cluster

To use OAuth 2.0 authentication in the Kafka cluster, you specify, for example, a tls listener configuration for your Kafka cluster custom resource with the authentication method oauth:

Assigining the authentication method type for OAuth 2.0

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    # ...
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: oauth
      #...

You can configure OAuth 2.0 authentication in your listeners. We recommend using OAuth 2.0 authentication together with TLS encryption (tls: true). Without encryption, the connection is vulnerable to network eavesdropping and unauthorized access through token theft.

You configure an external listener with type: oauth for a secure transport layer to communicate with the client.

Using OAuth 2.0 with an external listener

# ...
listeners:
  - name: external
    port: 9094
    type: loadbalancer
    tls: true
    authentication:
      type: oauth
    #...

The tls property is false by default, so it must be enabled.

When you have defined the type of authentication as OAuth 2.0, you add configuration based on the type of validation, either as fast local JWT validation or token validation using an introspection endpoint.

The procedure to configure OAuth 2.0 for listeners, with descriptions and examples, is described in Configuring OAuth 2.0 support for Kafka brokers.

6.4.2.3. Fast local JWT token validation configuration

Fast local JWT token validation checks a JWT token signature locally.

The local check ensures that a token:

Conforms to type by containing a (typ) claim value of Bearer for an access token
Is valid (not expired)
Has an issuer that matches a validIssuerURI

You specify a validIssuerURI attribute when you configure the listener, so that any tokens not issued by the authorization server are rejected.

The authorization server does not need to be contacted during fast local JWT token validation. You activate fast local JWT token validation by specifying a jwksEndpointUri attribute, the endpoint exposed by the OAuth 2.0 authorization server. The endpoint contains the public keys used to validate signed JWT tokens, which are sent as credentials by Kafka clients.

Note

All communication with the authorization server should be performed using TLS encryption.

You can configure a certificate truststore as an OpenShift Secret in your AMQ Streams project namespace, and use a tlsTrustedCertificates attribute to point to the OpenShift Secret containing the truststore file.

You might want to configure a userNameClaim to properly extract a username from the JWT token. If you want to use Kafka ACL authorization, you need to identify the user by their username during authentication. (The sub claim in JWT tokens is typically a unique ID, not a username.)

Example configuration for fast local JWT token validation

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    #...
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: oauth
          validIssuerUri: <https://<auth-server-address>/auth/realms/tls>
          jwksEndpointUri: <https://<auth-server-address>/auth/realms/tls/protocol/openid-connect/certs>
          userNameClaim: preferred_username
          maxSecondsWithoutReauthentication: 3600
          tlsTrustedCertificates:
          - secretName: oauth-server-cert
            certificate: ca.crt
    #...

6.4.2.4. OAuth 2.0 introspection endpoint configuration

Token validation using an OAuth 2.0 introspection endpoint treats a received access token as opaque. The Kafka broker sends an access token to the introspection endpoint, which responds with the token information necessary for validation. Importantly, it returns up-to-date information if the specific access token is valid, and also information about when the token expires.

To configure OAuth 2.0 introspection-based validation, you specify an introspectionEndpointUri attribute rather than the jwksEndpointUri attribute specified for fast local JWT token validation. Depending on the authorization server, you typically have to specify a clientId and clientSecret, because the introspection endpoint is usually protected.

Example configuration for an introspection endpoint

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: oauth
          clientId: kafka-broker
          clientSecret:
            secretName: my-cluster-oauth
            key: clientSecret
          validIssuerUri: <https://<auth-server-address>/auth/realms/tls>
          introspectionEndpointUri: <https://<auth-server-address>/auth/realms/tls/protocol/openid-connect/token/introspect>
          userNameClaim: preferred_username
          maxSecondsWithoutReauthentication: 3600
          tlsTrustedCertificates:
          - secretName: oauth-server-cert
            certificate: ca.crt

6.4.3. Session re-authentication for Kafka brokers

You can configure oauth listeners to use Kafka session re-authentication for OAuth 2.0 sessions between Kafka clients and Kafka brokers. This mechanism enforces the expiry of an authenticated session between the client and the broker after a defined period of time. When a session expires, the client immediately starts a new session by reusing the existing connection rather than dropping it.

Session re-authentication is disabled by default. To enable it, you set a time value for maxSecondsWithoutReauthentication in the oauth listener configuration. The same property is used to configure session re-authentication for OAUTHBEARER and PLAIN authentication. For an example configuration, see Section 6.4.6.2, “Configuring OAuth 2.0 support for Kafka brokers”.

Session re-authentication must be supported by the Kafka client libraries used by the client.

Session re-authentication can be used with fast local JWT or introspection endpoint token validation.

Client re-authentication

When the broker’s authenticated session expires, the client must re-authenticate to the existing session by sending a new, valid access token to the broker, without dropping the connection.

If token validation is successful, a new client session is started using the existing connection. If the client fails to re-authenticate, the broker will close the connection if further attempts are made to send or receive messages. Java clients that use Kafka client library 2.2 or later automatically re-authenticate if the re-authentication mechanism is enabled on the broker.

Session re-authentication also applies to refresh tokens, if used. When the session expires, the client refreshes the access token by using its refresh token. The client then uses the new access token to re-authenticate to the existing session.

Session expiry for OAUTHBEARER and PLAIN

When session re-authentication is configured, session expiry works differently for OAUTHBEARER and PLAIN authentication.

For OAUTHBEARER and PLAIN, using the client ID and secret method:

The broker’s authenticated session will expire at the configured maxSecondsWithoutReauthentication.
The session will expire earlier if the access token expires before the configured time.

For PLAIN using the long-lived access token method:

The broker’s authenticated session will expire at the configured maxSecondsWithoutReauthentication.
Re-authentication will fail if the access token expires before the configured time. Although session re-authentication is attempted, PLAIN has no mechanism for refreshing tokens.

If maxSecondsWithoutReauthentication is not configured, OAUTHBEARER and PLAIN clients can remain connected to brokers indefinitely, without needing to re-authenticate. Authenticated sessions do not end with access token expiry. However, this can be considered when configuring authorization, for example, by using keycloak authorization or installing a custom authorizer.

Additional resources

Section 6.4.2, “OAuth 2.0 Kafka broker configuration”
Section 6.4.6.2, “Configuring OAuth 2.0 support for Kafka brokers”
KafkaListenerAuthenticationOAuth schema reference
KIP-368

6.4.4. OAuth 2.0 Kafka client configuration

A Kafka client is configured with either:

The credentials required to obtain a valid access token from an authorization server (client ID and Secret)
A valid long-lived access token or refresh token, obtained using tools provided by an authorization server

The only information ever sent to the Kafka broker is an access token. The credentials used to authenticate with the authorization server to obtain the access token are never sent to the broker.

When a client obtains an access token, no further communication with the authorization server is needed.

The simplest mechanism is authentication with a client ID and Secret. Using a long-lived access token, or a long-lived refresh token, adds more complexity because there is an additional dependency on authorization server tools.

Note

If you are using long-lived access tokens, you may need to configure the client in the authorization server to increase the maximum lifetime of the token.

If the Kafka client is not configured with an access token directly, the client exchanges credentials for an access token during Kafka session initiation by contacting the authorization server. The Kafka client exchanges either:

Client ID and Secret
Client ID, refresh token, and (optionally) a secret
Username and password, with client ID and (optionally) a secret

6.4.5. OAuth 2.0 client authentication flows

OAuth 2.0 authentication flows depend on the underlying Kafka client and Kafka broker configuration. The flows must also be supported by the authorization server used.

The Kafka broker listener configuration determines how clients authenticate using an access token. The client can pass a client ID and secret to request an access token.

If a listener is configured to use PLAIN authentication, the client can authenticate with a client ID and secret or username and access token. These values are passed as the username and password properties of the PLAIN mechanism.

Listener configuration supports the following token validation options:

You can use fast local token validation based on JWT signature checking and local token introspection, without contacting an authorization server. The authorization server provides a JWKS endpoint with public certificates that are used to validate signatures on the tokens.
You can use a call to a token introspection endpoint provided by an authorization server. Each time a new Kafka broker connection is established, the broker passes the access token received from the client to the authorization server. The Kafka broker checks the response to confirm whether or not the token is valid.

Note

An authorization server might only allow the use of opaque access tokens, which means that local token validation is not possible.

Kafka client credentials can also be configured for the following types of authentication:

Direct local access using a previously generated long-lived access token
Contact with the authorization server for a new access token to be issued (using a client ID and a secret, or a refresh token, or a username and a password)

6.4.5.1. Example client authentication flows using the SASL OAUTHBEARER mechanism

You can use the following communication flows for Kafka authentication using the SASL OAUTHBEARER mechanism.

Client using client ID and secret, with broker delegating validation to authorization server
Client using client ID and secret, with broker performing fast local token validation
Client using long-lived access token, with broker delegating validation to authorization server
Client using long-lived access token, with broker performing fast local validation

Client using client ID and secret, with broker delegating validation to authorization server

Client using client ID and secret with broker delegating validation to authorization server

The Kafka client requests an access token from the authorization server using a client ID and secret, and optionally a refresh token. Alternatively, the client may authenticate using a username and a password.
The authorization server generates a new access token.
The Kafka client authenticates with the Kafka broker using the SASL OAUTHBEARER mechanism to pass the access token.
The Kafka broker validates the access token by calling a token introspection endpoint on the authorization server using its own client ID and secret.
A Kafka client session is established if the token is valid.

Client using client ID and secret, with broker performing fast local token validation

Client using client ID and secret with broker performing fast local token validation

The Kafka client authenticates with the authorization server from the token endpoint, using a client ID and secret, and optionally a refresh token. Alternatively, the client may authenticate using a username and a password.
The authorization server generates a new access token.
The Kafka client authenticates with the Kafka broker using the SASL OAUTHBEARER mechanism to pass the access token.
The Kafka broker validates the access token locally using a JWT token signature check, and local token introspection.

Client using long-lived access token, with broker delegating validation to authorization server

Client using long-lived access token with broker delegating validation to authorization server

The Kafka client authenticates with the Kafka broker using the SASL OAUTHBEARER mechanism to pass the long-lived access token.
The Kafka broker validates the access token by calling a token introspection endpoint on the authorization server, using its own client ID and secret.
A Kafka client session is established if the token is valid.

Client using long-lived access token, with broker performing fast local validation

Client using long-lived access token with broker performing fast local validation

The Kafka client authenticates with the Kafka broker using the SASL OAUTHBEARER mechanism to pass the long-lived access token.
The Kafka broker validates the access token locally using a JWT token signature check and local token introspection.

Warning

Fast local JWT token signature validation is suitable only for short-lived tokens as there is no check with the authorization server if a token has been revoked. Token expiration is written into the token, but revocation can happen at any time, so cannot be accounted for without contacting the authorization server. Any issued token would be considered valid until it expires.

6.4.5.2. Example client authentication flows using the SASL PLAIN mechanism

You can use the following communication flows for Kafka authentication using the OAuth PLAIN mechanism.

Client using a client ID and secret, with the broker obtaining the access token for the client
Client using a long-lived access token without a client ID and secret

Client using a client ID and secret, with the broker obtaining the access token for the client

Client using a client ID and secret with the broker obtaining the access token for the client

The Kafka client passes a clientId as a username and a secret as a password.
The Kafka broker uses a token endpoint to pass the clientId and secret to the authorization server.
The authorization server returns a fresh access token or an error if the client credentials are not valid.
The Kafka broker validates the token in one of the following ways:
1. If a token introspection endpoint is specified, the Kafka broker validates the access token by calling the endpoint on the authorization server. A session is established if the token validation is successful.
2. If local token introspection is used, a request is not made to the authorization server. The Kafka broker validates the access token locally using a JWT token signature check.

Client using a long-lived access token without a client ID and secret

Client using a long-lived access token without a client ID and secret

The Kafka client passes a username and password. The password provides the value of an access token that was obtained manually and configured before running the client.
The password is passed with or without an $accessToken: string prefix depending on whether or not the Kafka broker listener is configured with a token endpoint for authentication.
1. If the token endpoint is configured, the password should be prefixed by $accessToken: to let the broker know that the password parameter contains an access token rather than a client secret. The Kafka broker interprets the username as the account username.
2. If the token endpoint is not configured on the Kafka broker listener (enforcing a no-client-credentials mode), the password should provide the access token without the prefix. The Kafka broker interprets the username as the account username. In this mode, the client doesn’t use a client ID and secret, and the password parameter is always interpreted as a raw access token.
The Kafka broker validates the token in one of the following ways:
1. If a token introspection endpoint is specified, the Kafka broker validates the access token by calling the endpoint on the authorization server. A session is established if token validation is successful.
2. If local token introspection is used, there is no request made to the authorization server. Kafka broker validates the access token locally using a JWT token signature check.

6.4.6. Configuring OAuth 2.0 authentication

OAuth 2.0 is used for interaction between Kafka clients and AMQ Streams components.

In order to use OAuth 2.0 for AMQ Streams, you must:

Deploy an authorization server and configure the deployment to integrate with AMQ Streams
Deploy or update the Kafka cluster with Kafka broker listeners configured to use OAuth 2.0
Update your Java-based Kafka clients to use OAuth 2.0
Update Kafka component clients to use OAuth 2.0

6.4.6.1. Configuring an OAuth 2.0 authorization server

This procedure describes in general what you need to do to configure an authorization server for integration with AMQ Streams.

These instructions are not product specific.

The steps are dependent on the chosen authorization server. Consult the product documentation for the authorization server for information on how to set up OAuth 2.0 access.

Note

If you already have an authorization server deployed, you can skip the deployment step and use your current deployment.

Procedure

Deploy the authorization server to your cluster.
Access the CLI or admin console for the authorization server to configure OAuth 2.0 for AMQ Streams.
Now prepare the authorization server to work with AMQ Streams.
Configure a kafka-broker client.
Configure clients for each Kafka client component of your application.

What to do next

After deploying and configuring the authorization server, configure the Kafka brokers to use OAuth 2.0.

6.4.6.2. Configuring OAuth 2.0 support for Kafka brokers

This procedure describes how to configure Kafka brokers so that the broker listeners are enabled to use OAuth 2.0 authentication using an authorization server.

We advise use of OAuth 2.0 over an encrypted interface through through a listener with tls: true. Plain listeners are not recommended.

If the authorization server is using certificates signed by the trusted CA and matching the OAuth 2.0 server hostname, TLS connection works using the default settings. Otherwise, you may need to configure the truststore with proper certificates or disable the certificate hostname validation.

When configuring the Kafka broker you have two options for the mechanism used to validate the access token during OAuth 2.0 authentication of the newly connected Kafka client:

Configuring fast local JWT token validation
Configuring token validation using an introspection endpoint

Before you start

For more information on the configuration of OAuth 2.0 authentication for Kafka broker listeners, see:

KafkaListenerAuthenticationOAuth schema reference
OAuth 2.0 authentication mechanisms

Prerequisites

AMQ Streams and Kafka are running
An OAuth 2.0 authorization server is deployed

Procedure

Update the Kafka broker configuration (Kafka.spec.kafka) of your Kafka resource in an editor.
```
oc edit kafka my-cluster
```
Configure the Kafka broker listeners configuration.
The configuration for each type of listener does not have to be the same, as they are independent.
The examples here show the configuration options as configured for external listeners.
Example 1: Configuring fast local JWT token validation
```
#...
- name: external
  port: 9094
  type: loadbalancer
  tls: true
  authentication:
    type: oauth 1
    validIssuerUri: <https://<auth-server-address>/auth/realms/external> 2
    jwksEndpointUri: <https://<auth-server-address>/auth/realms/external/protocol/openid-connect/certs> 3
    userNameClaim: preferred_username 4
    maxSecondsWithoutReauthentication: 3600 5
    tlsTrustedCertificates: 6
    - secretName: oauth-server-cert
      certificate: ca.crt
    disableTlsHostnameVerification: true 7
    jwksExpirySeconds: 360 8
    jwksRefreshSeconds: 300 9
    jwksMinRefreshPauseSeconds: 1 10
```
1
Listener type set to oauth.
2
URI of the token issuer used for authentication.
3
URI of the JWKS certificate endpoint used for local JWT validation.
4
The token claim (or key) that contains the actual user name in the token. The user name is the principal used to identify the user. The userNameClaim value will depend on the authentication flow and the authorization server used.
5
(Optional) Activates the Kafka re-authentication mechanism that enforces session expiry to the same length of time as the access token. If the specified value is less than the time left for the access token to expire, then the client will have to re-authenticate before the actual token expiry. By default, the session does not expire when the access token expires, and the client does not attempt re-authentication.
6
(Optional) Trusted certificates for TLS connection to the authorization server.
7
(Optional) Disable TLS hostname verification. Default is false.
8
The duration the JWKS certificates are considered valid before they expire. Default is 360 seconds. If you specify a longer time, consider the risk of allowing access to revoked certificates.
9
The period between refreshes of JWKS certificates. The interval must be at least 60 seconds shorter than the expiry interval. Default is 300 seconds.
10
The minimum pause in seconds between consecutive attempts to refresh JWKS public keys. When an unknown signing key is encountered, the JWKS keys refresh is scheduled outside the regular periodic schedule with at least the specified pause since the last refresh attempt. The refreshing of keys follows the rule of exponential backoff, retrying on unsuccessful refreshes with ever increasing pause, until it reaches jwksRefreshSeconds. The default value is 1.
Example 2: Configuring token validation using an introspection endpoint
```
- name: external
  port: 9094
  type: loadbalancer
  tls: true
  authentication:
    type: oauth
    validIssuerUri: <https://<auth-server-address>/auth/realms/external>
    introspectionEndpointUri: <https://<auth-server-address>/auth/realms/external/protocol/openid-connect/token/introspect> 1
    clientId: kafka-broker 2
    clientSecret: 3
      secretName: my-cluster-oauth
      key: clientSecret
    userNameClaim: preferred_username 4
    maxSecondsWithoutReauthentication: 3600 5
```
1
URI of the token introspection endpoint.
2
Client ID to identify the client.
3
Client Secret and client ID is used for authentication.
4
The token claim (or key) that contains the actual user name in the token. The user name is the principal used to identify the user. The userNameClaim value will depend on the authorization server used.
5
(Optional) Activates the Kafka re-authentication mechanism that enforces session expiry to the same length of time as the access token. If the specified value is less than the time left for the access token to expire, then the client will have to re-authenticate before the actual token expiry. By default, the session does not expire when the access token expires, and the client does not attempt re-authentication.
Depending on how you apply OAuth 2.0 authentication, and the type of authorization server, there are additional (optional) configuration settings you can use:
```
  # ...
  authentication:
    type: oauth
    # ...
    checkIssuer: false 1
    checkAudience: true 2
    fallbackUserNameClaim: client_id 3
    fallbackUserNamePrefix: client-account- 4
    validTokenType: bearer 5
    userInfoEndpointUri: https://OAUTH-SERVER-ADDRESS/auth/realms/external/protocol/openid-connect/userinfo 6
    enableOauthBearer: false 7
    enablePlain: true 8
    tokenEndpointUri: https://OAUTH-SERVER-ADDRESS/auth/realms/external/protocol/openid-connect/token 9
    customClaimCheck: "@.custom == 'custom-value'" 10
    clientAudience: AUDIENCE 11
    clientScope: SCOPE 12
    connectTimeoutSeconds: 60 13
    readTimeoutSeconds: 60 14
    groupsClaim: "$.groups" 15
    groupsClaimDelimiter: "," 16
```
1
If your authorization server does not provide an iss claim, it is not possible to perform an issuer check. In this situation, set checkIssuer to false and do not specify a validIssuerUri. Default is true.
2
If your authorization server provides an aud (audience) claim, and you want to enforce an audience check, set checkAudience to true. Audience checks identify the intended recipients of tokens. As a result, the Kafka broker will reject tokens that do not have its clientId in their aud claim. Default is false.
3
An authorization server may not provide a single attribute to identify both regular users and clients. When a client authenticates in its own name, the server might provide a client ID. When a user authenticates using a username and password, to obtain a refresh token or an access token, the server might provide a username attribute in addition to a client ID. Use this fallback option to specify the username claim (attribute) to use if a primary user ID attribute is not available.
4
In situations where fallbackUserNameClaim is applicable, it may also be necessary to prevent name collisions between the values of the username claim, and those of the fallback username claim. Consider a situation where a client called producer exists, but also a regular user called producer exists. In order to differentiate between the two, you can use this property to add a prefix to the user ID of the client.
5
(Only applicable when using introspectionEndpointUri) Depending on the authorization server you are using, the introspection endpoint may or may not return the token type attribute, or it may contain different values. You can specify a valid token type value that the response from the introspection endpoint has to contain.
6
(Only applicable when using introspectionEndpointUri) The authorization server may be configured or implemented in such a way to not provide any identifiable information in an Introspection Endpoint response. In order to obtain the user ID, you can configure the URI of the userinfo endpoint as a fallback. The userNameClaim, fallbackUserNameClaim, and fallbackUserNamePrefix settings are applied to the response of userinfo endpoint.
7
Set this to false to disable the OAUTHBEARER mechanism on the listener. At least one of PLAIN or OAUTHBEARER has to be enabled. Default is true.
8
Set to true to enable PLAIN authentication on the listener, which is supported for clients on all platforms.
9
Additional configuration for the PLAIN mechanism. If specified, clients can authenticate over PLAIN by passing an access token as the password using an $accessToken: prefix. For production, always use https:// urls.
10
Additional custom rules can be imposed on the JWT access token during validation by setting this to a JsonPath filter query. If the access token does not contain the necessary data, it is rejected. When using the introspectionEndpointUri, the custom check is applied to the introspection endpoint response JSON.
11
An audience parameter passed to the token endpoint. An audience is used when obtaining an access token for inter-broker authentication. It is also used in the name of a client for OAuth 2.0 over PLAIN client authentication using a clientId and secret. This only affects the ability to obtain the token, and the content of the token, depending on the authorization server. It does not affect token validation rules by the listener.
12
A scope parameter passed to the token endpoint. A scope is used when obtaining an access token for inter-broker authentication. It is also used in the name of a client for OAuth 2.0 over PLAIN client authentication using a clientId and secret. This only affects the ability to obtain the token, and the content of the token, depending on the authorization server. It does not affect token validation rules by the listener.
13
The connect timeout in seconds when connecting to the authorization server. The default value is 60.
14
The read timeout in seconds when connecting to the authorization server. The default value is 60.
15
A JsonPath query used to extract groups information from JWT token or introspection endpoint response. Not set by default. This can be used by a custom authorizer to make authorization decisions based on user groups.
16
A delimiter used to parse groups information when returned as a single delimited string. The default value is ',' (comma).
Save and exit the editor, then wait for rolling updates to complete.
Check the update in the logs or by watching the pod state transitions:
```
oc logs -f ${POD_NAME} -c ${CONTAINER_NAME}
oc get pod -w
```
The rolling update configures the brokers to use OAuth 2.0 authentication.

What to do next

Configure your Kafka clients to use OAuth 2.0

6.4.6.3. Configuring Kafka Java clients to use OAuth 2.0

Configure Kafka producer and consumer APIs to use OAuth 2.0 for interaction with Kafka brokers. Add a callback plugin to your client pom.xml file, then configure your client for OAuth 2.0.

Specify the following in your client configuration:

A SASL (Simple Authentication and Security Layer) security protocol:
- SASL_SSL for authentication over TLS encrypted connections
- SASL_PLAINTEXT for authentication over unencrypted connections
  Use SASL_SSL for production and SASL_PLAINTEXT for local development only. When using SASL_SSL, additional ssl.truststore configuration is needed. The truststore configuration is required for secure connection (https://) to the OAuth 2.0 authorization server. To verify the OAuth 2.0 authorization server, add the CA certificate for the authorization server to the truststore in your client configuration. You can configure a truststore in PEM or PKCS #12 format.
A Kafka SASL mechanism:
- OAUTHBEARER for credentials exchange using a bearer token
- PLAIN to pass client credentials (clientId + secret) or an access token
A JAAS (Java Authentication and Authorization Service) module that implements the SASL mechanism:
- org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule implements the OAUTHBEARER mechanism
- org.apache.kafka.common.security.plain.PlainLoginModule implements the PLAIN mechanism
SASL authentication properties, which support the following authentication methods:
- OAuth 2.0 client credentials
- OAuth 2.0 password grant (deprecated)
- Access token
- Refresh token

Add the SASL authentication properties as JAAS configuration (sasl.jaas.config). How you configure the authentication properties depends on the authentication method you are using to access the OAuth 2.0 authorization server. In this procedure, the properties are specified in a properties file, then loaded into the client configuration.

Note

You can also specify authentication properties as environment variables, or as Java system properties. For Java system properties, you can set them using setProperty and pass them on the command line using the -D option.

Prerequisites

AMQ Streams and Kafka are running
An OAuth 2.0 authorization server is deployed and configured for OAuth access to Kafka brokers
Kafka brokers are configured for OAuth 2.0

Procedure

Add the client library with OAuth 2.0 support to the pom.xml file for the Kafka client:

<dependency>
 <groupId>io.strimzi</groupId>
 <artifactId>kafka-oauth-client</artifactId>
 <version>0.11.0.redhat-00003</version>
</dependency>

Configure the client properties by specifying the following configuration in a properties file:

The security protocol
The SASL mechanism

The JAAS module and authentication properties according to the method being used

For example, we can add the following to a client.properties file:

Client credentials mechanism properties

security.protocol=SASL_SSL 1
sasl.mechanism=OAUTHBEARER 2
ssl.truststore.location=/tmp/truststore.p12 3
ssl.truststore.password=$STOREPASS
ssl.truststore.type=PKCS12
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.token.endpoint.uri="<token_endpoint_url>" \ 4
  oauth.client.id="<client_id>" \ 5
  oauth.client.secret="<client_secret>" \ 6
  oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \ 7
  oauth.ssl.truststore.password="$STOREPASS" \ 8
  oauth.ssl.truststore.type="PKCS12" \ 9
  oauth.scope="<scope>" \ 10
  oauth.audience="<audience>" ; 11

1: SASL_SSL security protocol for TLS-encrypted connections. Use SASL_PLAINTEXT over unencrypted connections for local development only.
2: The SASL mechanism specified as OAUTHBEARER or PLAIN.
3: The truststore configuration for secure access to the Kafka cluster.
4: URI of the authorization server token endpoint.
5: Client ID, which is the name used when creating the client in the authorization server.
6: Client secret created when creating the client in the authorization server.
7: The location contains the public key certificate (truststore.p12) for the authorization server.
8: The password for accessing the truststore.
9: The truststore type.
10: (Optional) The scope for requesting the token from the token endpoint. An authorization server may require a client to specify the scope.
11: (Optional) The audience for requesting the token from the token endpoint. An authorization server may require a client to specify the audience.

Password grants mechanism properties

security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
ssl.truststore.location=/tmp/truststore.p12
ssl.truststore.password=$STOREPASS
ssl.truststore.type=PKCS12
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.ssl.truststore.location="/tmp/oauth-truststore.p12" \
  oauth.ssl.truststore.password="$STOREPASS" \
  oauth.ssl.truststore.type="PKCS12" \
  oauth.scope="<scope>" \
  oauth.audience="<audience>" ;

1: Client ID, which is the name used when creating the client in the authorization server.
2: (Optional) Client secret created when creating the client in the authorization server.
3: Username for password grant authentication. OAuth password grant configuration (username and password) uses the OAuth 2.0 password grant method. To use password grants, create a user account for a client on your authorization server with limited permissions. The account should act like a service account. Use in environments where user accounts are required for authentication, but consider using a refresh token first.
4: Password for password grant authentication.
Note
SASL PLAIN does not support passing a username and password (password grants) using the OAuth 2.0 password grant method.

Access token properties

security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
ssl.truststore.location=/tmp/truststore.p12
ssl.truststore.password=$STOREPASS
ssl.truststore.type=PKCS12
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.token.endpoint.uri="<token_endpoint_url>" \
  oauth.access.token="<access_token>" ; 1
  oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
  oauth.ssl.truststore.password="$STOREPASS" \
  oauth.ssl.truststore.type="PKCS12" \

1: Long-lived access token for Kafka clients.

Refresh token properties

security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
ssl.truststore.location=/tmp/truststore.p12
ssl.truststore.password=$STOREPASS
ssl.truststore.type=PKCS12
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.refresh.token="<refresh_token>" ; 3
  oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
  oauth.ssl.truststore.password="$STOREPASS" \
  oauth.ssl.truststore.type="PKCS12" \

1: Client ID, which is the name used when creating the client in the authorization server.
2: (Optional) Client secret created when creating the client in the authorization server.
3: Long-lived refresh token for Kafka clients.

Input the client properties for OAUTH 2.0 authentication into the Java client code.

Example showing input of client properties

Properties props = new Properties();
try (FileReader reader = new FileReader("client.properties", StandardCharsets.UTF_8)) {
  props.load(reader);
}

Verify that the Kafka client can access the Kafka brokers.

6.4.6.4. Configuring OAuth 2.0 for Kafka components

This procedure describes how to configure Kafka components to use OAuth 2.0 authentication using an authorization server.

You can configure authentication for:

Kafka Connect
Kafka MirrorMaker
Kafka Bridge

In this scenario, the Kafka component and the authorization server are running in the same cluster.

Before you start

For more information on the configuration of OAuth 2.0 authentication for Kafka components, see:

KafkaClientAuthenticationOAuth schema reference

Prerequisites

AMQ Streams and Kafka are running
An OAuth 2.0 authorization server is deployed and configured for OAuth access to Kafka brokers
Kafka brokers are configured for OAuth 2.0

Procedure

Create a client secret and mount it to the component as an environment variable.
For example, here we are creating a client Secret for the Kafka Bridge:
```
apiVersion: kafka.strimzi.io/v1beta2
kind: Secret
metadata:
 name: my-bridge-oauth
type: Opaque
data:
 clientSecret: MGQ1OTRmMzYtZTllZS00MDY2LWI5OGEtMTM5MzM2NjdlZjQw 1
```
1
The clientSecret key must be in base64 format.
Create or edit the resource for the Kafka component so that OAuth 2.0 authentication is configured for the authentication property.
For OAuth 2.0 authentication, you can use:
- Client ID and secret
- Client ID and refresh token
- Access token
- Username and password
- TLS
KafkaClientAuthenticationOAuth schema reference provides examples of each.
For example, here OAuth 2.0 is assigned to the Kafka Bridge client using a client ID and secret, and TLS:
```
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaBridge
metadata:
  name: my-bridge
spec:
  # ...
  authentication:
    type: oauth 1
    tokenEndpointUri: https://<auth-server-address>/auth/realms/master/protocol/openid-connect/token 2
    clientId: kafka-bridge
    clientSecret:
      secretName: my-bridge-oauth
      key: clientSecret
    tlsTrustedCertificates: 3
    - secretName: oauth-server-cert
      certificate: tls.crt
```
1
Authentication type set to oauth.
2
URI of the token endpoint for authentication.
3
Trusted certificates for TLS connection to the authorization server.
Depending on how you apply OAuth 2.0 authentication, and the type of authorization server, there are additional configuration options you can use:
```
# ...
spec:
  # ...
  authentication:
    # ...
    disableTlsHostnameVerification: true 1
    checkAccessTokenType: false 2
    accessTokenIsJwt: false 3
    scope: any 4
    audience: kafka 5
    connectTimeoutSeconds: 60 6
    readTimeoutSeconds: 60 7
```
1
(Optional) Disable TLS hostname verification. Default is false.
2
If the authorization server does not return a typ (type) claim inside the JWT token, you can apply checkAccessTokenType: false to skip the token type check. Default is true.
3
If you are using opaque tokens, you can apply accessTokenIsJwt: false so that access tokens are not treated as JWT tokens.
4
(Optional) The scope for requesting the token from the token endpoint. An authorization server may require a client to specify the scope. In this case it is any.
5
(Optional) The audience for requesting the token from the token endpoint. An authorization server may require a client to specify the audience. In this case it is kafka.
6
(Optional) The connect timeout in seconds when connecting to the authorization server. The default value is 60.
7
(Optional) The read timeout in seconds when connecting to the authorization server. The default value is 60.
Apply the changes to the deployment of your Kafka resource.
```
oc apply -f your-file
```
Check the update in the logs or by watching the pod state transitions:
```
oc logs -f ${POD_NAME} -c ${CONTAINER_NAME}
oc get pod -w
```
The rolling updates configure the component for interaction with Kafka brokers using OAuth 2.0 authentication.

6.5. Using OAuth 2.0 token-based authorization

If you are using OAuth 2.0 with Red Hat Single Sign-On for token-based authentication, you can also use Red Hat Single Sign-On to configure authorization rules to constrain client access to Kafka brokers. Authentication establishes the identity of a user. Authorization decides the level of access for that user.

AMQ Streams supports the use of OAuth 2.0 token-based authorization through Red Hat Single Sign-On Authorization Services, which allows you to manage security policies and permissions centrally.

Security policies and permissions defined in Red Hat Single Sign-On are used to grant access to resources on Kafka brokers. Users and clients are matched against policies that permit access to perform specific actions on Kafka brokers.

Kafka allows all users full access to brokers by default, and also provides the AclAuthorizer plugin to configure authorization based on Access Control Lists (ACLs).

ZooKeeper stores ACL rules that grant or deny access to resources based on username. However, OAuth 2.0 token-based authorization with Red Hat Single Sign-On offers far greater flexibility on how you wish to implement access control to Kafka brokers. In addition, you can configure your Kafka brokers to use OAuth 2.0 authorization and ACLs.

Additional resources

Using OAuth 2.0 token-based authentication
Kafka Authorization
Red Hat Single Sign-On documentation

6.5.1. OAuth 2.0 authorization mechanism

OAuth 2.0 authorization in AMQ Streams uses Red Hat Single Sign-On server Authorization Services REST endpoints to extend token-based authentication with Red Hat Single Sign-On by applying defined security policies on a particular user, and providing a list of permissions granted on different resources for that user. Policies use roles and groups to match permissions to users. OAuth 2.0 authorization enforces permissions locally based on the received list of grants for the user from Red Hat Single Sign-On Authorization Services.

6.5.1.1. Kafka broker custom authorizer

A Red Hat Single Sign-On authorizer (KeycloakRBACAuthorizer) is provided with AMQ Streams. To be able to use the Red Hat Single Sign-On REST endpoints for Authorization Services provided by Red Hat Single Sign-On, you configure a custom authorizer on the Kafka broker.

The authorizer fetches a list of granted permissions from the authorization server as needed, and enforces authorization locally on the Kafka Broker, making rapid authorization decisions for each client request.

6.5.2. Configuring OAuth 2.0 authorization support

This procedure describes how to configure Kafka brokers to use OAuth 2.0 authorization using Red Hat Single Sign-On Authorization Services.

Before you begin

Consider the access you require or want to limit for certain users. You can use a combination of Red Hat Single Sign-On groups, roles, clients, and users to configure access in Red Hat Single Sign-On.

Typically, groups are used to match users based on organizational departments or geographical locations. And roles are used to match users based on their function.

With Red Hat Single Sign-On, you can store users and groups in LDAP, whereas clients and roles cannot be stored this way. Storage and access to user data may be a factor in how you choose to configure authorization policies.

Note

Super users always have unconstrained access to a Kafka broker regardless of the authorization implemented on the Kafka broker.

Prerequisites

AMQ Streams must be configured to use OAuth 2.0 with Red Hat Single Sign-On for token-based authentication. You use the same Red Hat Single Sign-On server endpoint when you set up authorization.
OAuth 2.0 authentication must be configured with the maxSecondsWithoutReauthentication option to enable re-authentication.

Procedure

Access the Red Hat Single Sign-On Admin Console or use the Red Hat Single Sign-On Admin CLI to enable Authorization Services for the Kafka broker client you created when setting up OAuth 2.0 authentication.
Use Authorization Services to define resources, authorization scopes, policies, and permissions for the client.
Bind the permissions to users and clients by assigning them roles and groups.
Configure the Kafka brokers to use Red Hat Single Sign-On authorization by updating the Kafka broker configuration (Kafka.spec.kafka) of your Kafka resource in an editor.
```
oc edit kafka my-cluster
```
Configure the Kafka broker kafka configuration to use keycloak authorization, and to be able to access the authorization server and Authorization Services.
For example:
```
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    authorization:
      type: keycloak 1
      tokenEndpointUri: <https://<auth-server-address>/auth/realms/external/protocol/openid-connect/token> 2
      clientId: kafka 3
      delegateToKafkaAcls: false 4
      disableTlsHostnameVerification: false 5
      superUsers: 6
      - CN=fred
      - sam
      - CN=edward
      tlsTrustedCertificates: 7
      - secretName: oauth-server-cert
        certificate: ca.crt
      grantsRefreshPeriodSeconds: 60 8
      grantsRefreshPoolSize: 5 9
      connectTimeoutSeconds: 60 10
      readTimeoutSeconds: 60 11
    #...
```
1
Type keycloak enables Red Hat Single Sign-On authorization.
2
URI of the Red Hat Single Sign-On token endpoint. For production, always use https:// urls. When you configure token-based oauth authentication, you specify a jwksEndpointUri as the URI for local JWT validation. The hostname for the tokenEndpointUri URI must be the same.
3
The client ID of the OAuth 2.0 client definition in Red Hat Single Sign-On that has Authorization Services enabled. Typically, kafka is used as the ID.
4
(Optional) Delegate authorization to Kafka AclAuthorizer if access is denied by Red Hat Single Sign-On Authorization Services policies. Default is false.
5
(Optional) Disable TLS hostname verification. Default is false.
6
(Optional) Designated super users.
7
(Optional) Trusted certificates for TLS connection to the authorization server.
8
(Optional) The time between two consecutive grants refresh runs. That is the maximum time for active sessions to detect any permissions changes for the user on Red Hat Single Sign-On. The default value is 60.
9
(Optional) The number of threads to use to refresh (in parallel) the grants for the active sessions. The default value is 5.
10
(Optional) The connect timeout in seconds when connecting to the Red Hat Single Sign-On token endpoint. The default value is 60.
11
(Optional) The read timeout in seconds when connecting to the Red Hat Single Sign-On token endpoint. The default value is 60.
Save and exit the editor, then wait for rolling updates to complete.
Check the update in the logs or by watching the pod state transitions:
```
oc logs -f ${POD_NAME} -c kafka
oc get pod -w
```
The rolling update configures the brokers to use OAuth 2.0 authorization.
Verify the configured permissions by accessing Kafka brokers as clients or users with specific roles, making sure they have the necessary access, or do not have the access they are not supposed to have.

6.5.3. Managing policies and permissions in Red Hat Single Sign-On Authorization Services

This section describes the authorization models used by Red Hat Single Sign-On Authorization Services and Kafka, and defines the important concepts in each model.

To grant permissions to access Kafka, you can map Red Hat Single Sign-On Authorization Services objects to Kafka resources by creating an OAuth client specification in Red Hat Single Sign-On. Kafka permissions are granted to user accounts or service accounts using Red Hat Single Sign-On Authorization Services rules.

Examples are shown of the different user permissions required for common Kafka operations, such as creating and listing topics.

6.5.3.1. Kafka and Red Hat Single Sign-On authorization models overview

Kafka and Red Hat Single Sign-On Authorization Services use different authorization models.

Kafka authorization model

Kafka’s authorization model uses resource types. When a Kafka client performs an action on a broker, the broker uses the configured KeycloakRBACAuthorizer to check the client’s permissions, based on the action and resource type.

Kafka uses five resource types to control access: Topic, Group, Cluster, TransactionalId, and DelegationToken. Each resource type has a set of available permissions.

Topic

Create
Write
Read
Delete
Describe
DescribeConfigs
Alter
AlterConfigs

Group

Read
Describe
Delete

Cluster

Create
Describe
Alter
DescribeConfigs
AlterConfigs
IdempotentWrite
ClusterAction

TransactionalId

Describe
Write

DelegationToken

Describe

Red Hat Single Sign-On Authorization Services model

The Red Hat Single Sign-On Authorization Services model has four concepts for defining and granting permissions: resources, authorization scopes, policies, and permissions.

Resources

A resource is a set of resource definitions that are used to match resources with permitted actions. A resource might be an individual topic, for example, or all topics with names starting with the same prefix. A resource definition is associated with a set of available authorization scopes, which represent a set of all actions available on the resource. Often, only a subset of these actions is actually permitted.

Authorization scopes

An authorization scope is a set of all the available actions on a specific resource definition. When you define a new resource, you add scopes from the set of all scopes.

Policies

A policy is an authorization rule that uses criteria to match against a list of accounts. Policies can match:

Service accounts based on client ID or roles
User accounts based on username, groups, or roles.

Permissions

A permission grants a subset of authorization scopes on a specific resource definition to a set of users.

Additional resources

Kafka authorization model

6.5.3.2. Map Red Hat Single Sign-On Authorization Services to the Kafka authorization model

The Kafka authorization model is used as a basis for defining the Red Hat Single Sign-On roles and resources that will control access to Kafka.

To grant Kafka permissions to user accounts or service accounts, you first create an OAuth client specification in Red Hat Single Sign-On for the Kafka broker. You then specify Red Hat Single Sign-On Authorization Services rules on the client. Typically, the client id of the OAuth client that represents the broker is kafka. The example configuration files provided with AMQ Streams use kafka as the OAuth client id.

Note

If you have multiple Kafka clusters, you can use a single OAuth client (kafka) for all of them. This gives you a single, unified space in which to define and manage authorization rules. However, you can also use different OAuth client ids (for example, my-cluster-kafka or cluster-dev-kafka) and define authorization rules for each cluster within each client configuration.

The kafka client definition must have the Authorization Enabled option enabled in the Red Hat Single Sign-On Admin Console.

All permissions exist within the scope of the kafka client. If you have different Kafka clusters configured with different OAuth client IDs, they each need a separate set of permissions even though they’re part of the same Red Hat Single Sign-On realm.

When the Kafka client uses OAUTHBEARER authentication, the Red Hat Single Sign-On authorizer (KeycloakRBACAuthorizer) uses the access token of the current session to retrieve a list of grants from the Red Hat Single Sign-On server. To retrieve the grants, the authorizer evaluates the Red Hat Single Sign-On Authorization Services policies and permissions.

Authorization scopes for Kafka permissions

An initial Red Hat Single Sign-On configuration usually involves uploading authorization scopes to create a list of all possible actions that can be performed on each Kafka resource type. This step is performed once only, before defining any permissions. You can add authorization scopes manually instead of uploading them.

Authorization scopes must contain all the possible Kafka permissions regardless of the resource type:

Create
Write
Read
Delete
Describe
Alter
DescribeConfig
AlterConfig
ClusterAction
IdempotentWrite

Note

If you’re certain you won’t need a permission (for example, IdempotentWrite), you can omit it from the list of authorization scopes. However, that permission won’t be available to target on Kafka resources.

Resource patterns for permissions checks

Resource patterns are used for pattern matching against the targeted resources when performing permission checks. The general pattern format is RESOURCE-TYPE:PATTERN-NAME.

The resource types mirror the Kafka authorization model. The pattern allows for two matching options:

Exact matching (when the pattern does not end with *)
Prefix matching (when the pattern ends with *)

Example patterns for resources

Topic:my-topic
Topic:orders-*
Group:orders-*
Cluster:*

Additionally, the general pattern format can be prefixed by kafka-cluster:CLUSTER-NAME followed by a comma, where CLUSTER-NAME refers to the metadata.name in the Kafka custom resource.

Example patterns for resources with cluster prefix

kafka-cluster:my-cluster,Topic:*
kafka-cluster:*,Group:b_*

When the kafka-cluster prefix is missing, it is assumed to be kafka-cluster:*.

When defining a resource, you can associate it with a list of possible authorization scopes which are relevant to the resource. Set whatever actions make sense for the targeted resource type.

Though you may add any authorization scope to any resource, only the scopes supported by the resource type are considered for access control.

Policies for applying access permission

Policies are used to target permissions to one or more user accounts or service accounts. Targeting can refer to:

Specific user or service accounts
Realm roles or client roles
User groups
JavaScript rules to match a client IP address

A policy is given a unique name and can be reused to target multiple permissions to multiple resources.

Permissions to grant access

Use fine-grained permissions to pull together the policies, resources, and authorization scopes that grant access to users.

The name of each permission should clearly define which permissions it grants to which users. For example, Dev Team B can read from topics starting with x.

Additional resources

For more information about how to configure permissions through Red Hat Single Sign-On Authorization Services, see Section 6.5.4, “Trying Red Hat Single Sign-On Authorization Services”.

6.5.3.3. Example permissions required for Kafka operations

The following examples demonstrate the user permissions required for performing common operations on Kafka.

Create a topic

To create a topic, the Create permission is required for the specific topic, or for Cluster:kafka-cluster.

bin/kafka-topics.sh --create --topic my-topic \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties

List topics

If a user has the Describe permission on a specified topic, the topic is listed.

bin/kafka-topics.sh --list \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties

Display topic details

To display a topic’s details, Describe and DescribeConfigs permissions are required on the topic.

bin/kafka-topics.sh --describe --topic my-topic \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties

Produce messages to a topic

To produce messages to a topic, Describe and Write permissions are required on the topic.

If the topic hasn’t been created yet, and topic auto-creation is enabled, the permissions to create a topic are required.

bin/kafka-console-producer.sh  --topic my-topic \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --producer.config=/tmp/config.properties

Consume messages from a topic

To consume messages from a topic, Describe and Read permissions are required on the topic. Consuming from the topic normally relies on storing the consumer offsets in a consumer group, which requires additional Describe and Read permissions on the consumer group.

Two resources are needed for matching. For example:

Topic:my-topic
Group:my-group-*

bin/kafka-console-consumer.sh --topic my-topic --group my-group-1 --from-beginning \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --consumer.config /tmp/config.properties

Produce messages to a topic using an idempotent producer

As well as the permissions for producing to a topic, an additional IdempotentWrite permission is required on the Cluster:kafka-cluster resource.

Two resources are needed for matching. For example:

Topic:my-topic
Cluster:kafka-cluster

bin/kafka-console-producer.sh  --topic my-topic \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --producer.config=/tmp/config.properties --producer-property enable.idempotence=true --request-required-acks -1

List consumer groups

When listing consumer groups, only the groups on which the user has the Describe permissions are returned. Alternatively, if the user has the Describe permission on the Cluster:kafka-cluster, all the consumer groups are returned.

bin/kafka-consumer-groups.sh --list \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties

Display consumer group details

To display a consumer group’s details, the Describe permission is required on the group and the topics associated with the group.

bin/kafka-consumer-groups.sh --describe --group my-group-1 \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties

Change topic configuration

To change a topic’s configuration, the Describe and Alter permissions are required on the topic.

bin/kafka-topics.sh --alter --topic my-topic --partitions 2 \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties

Display Kafka broker configuration

In order to use kafka-configs.sh to get a broker’s configuration, the DescribeConfigs permission is required on the Cluster:kafka-cluster.

bin/kafka-configs.sh --entity-type brokers --entity-name 0 --describe --all \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties

Change Kafka broker configuration

To change a Kafka broker’s configuration, DescribeConfigs and AlterConfigs permissions are required on Cluster:kafka-cluster.

bin/kafka-configs --entity-type brokers --entity-name 0 --alter --add-config log.cleaner.threads=2 \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties

Delete a topic

To delete a topic, the Describe and Delete permissions are required on the topic.

bin/kafka-topics.sh --delete --topic my-topic \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties

Select a lead partition

To run leader selection for topic partitions, the Alter permission is required on the Cluster:kafka-cluster.

bin/kafka-leader-election.sh --topic my-topic --partition 0 --election-type PREFERRED  /
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --admin.config /tmp/config.properties

Reassign partitions

To generate a partition reassignment file, Describe permissions are required on the topics involved.

bin/kafka-reassign-partitions.sh --topics-to-move-json-file /tmp/topics-to-move.json --broker-list "0,1" --generate \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config /tmp/config.properties > /tmp/partition-reassignment.json

To execute the partition reassignment, Describe and Alter permissions are required on Cluster:kafka-cluster. Also, Describe permissions are required on the topics involved.

bin/kafka-reassign-partitions.sh --reassignment-json-file /tmp/partition-reassignment.json --execute \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config /tmp/config.properties

To verify partition reassignment, Describe, and AlterConfigs permissions are required on Cluster:kafka-cluster, and on each of the topics involved.

bin/kafka-reassign-partitions.sh --reassignment-json-file /tmp/partition-reassignment.json --verify \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config /tmp/config.properties

6.5.4. Trying Red Hat Single Sign-On Authorization Services

This example explains how to use Red Hat Single Sign-On Authorization Services with keycloak authorization. Use Red Hat Single Sign-On Authorization Services to enforce access restrictions on Kafka clients. Red Hat Single Sign-On Authorization Services use authorization scopes, policies and permissions to define and apply access control to resources.

Red Hat Single Sign-On Authorization Services REST endpoints provide a list of granted permissions on resources for authenticated users. The list of grants (permissions) is fetched from the Red Hat Single Sign-On server as the first action after an authenticated session is established by the Kafka client. The list is refreshed in the background so that changes to the grants are detected. Grants are cached and enforced locally on the Kafka broker for each user session to provide fast authorization decisions.

AMQ Streams provides example configuration files. These include the following example files for setting up Red Hat Single Sign-On:

kafka-ephemeral-oauth-single-keycloak-authz.yaml: An example Kafka custom resource configured for OAuth 2.0 token-based authorization using Red Hat Single Sign-On. You can use the custom resource to deploy a Kafka cluster that uses keycloak authorization and token-based oauth authentication.
kafka-authz-realm.json: An example Red Hat Single Sign-On realm configured with sample groups, users, roles and clients. You can import the realm into a Red Hat Single Sign-On instance to set up fine-grained permissions to access Kafka.

If you want to try the example with Red Hat Single Sign-On, use these files to perform the tasks outlined in this section in the order shown.

Accessing the Red Hat Single Sign-On Admin Console
Deploying a Kafka cluster with Red Hat Single Sign-On authorization
Preparing TLS connectivity for a CLI Kafka client session
Checking authorized access to Kafka using a CLI Kafka client session

Authentication

When you configure token-based oauth authentication, you specify a jwksEndpointUri as the URI for local JWT validation. When you configure keycloak authorization, you specify a tokenEndpointUri as the URI of the Red Hat Single Sign-On token endpoint. The hostname for both URIs must be the same.

Targeted permissions with group or role policies

In Red Hat Single Sign-On, confidential clients with service accounts enabled can authenticate to the server in their own name using a client ID and a secret. This is convenient for microservices that typically act in their own name, and not as agents of a particular user (like a web site). Service accounts can have roles assigned like regular users. They cannot, however, have groups assigned. As a consequence, if you want to target permissions to microservices using service accounts, you cannot use group policies, and should instead use role policies. Conversely, if you want to limit certain permissions only to regular user accounts where authentication with a username and password is required, you can achieve that as a side effect of using the group policies rather than the role policies. This is what is used in this example for permissions that start with ClusterManager. Performing cluster management is usually done interactively using CLI tools. It makes sense to require the user to log in before using the resulting access token to authenticate to the Kafka broker. In this case, the access token represents the specific user, rather than the client application.

6.5.4.1. Accessing the Red Hat Single Sign-On Admin Console

Set up Red Hat Single Sign-On, then connect to its Admin Console and add the preconfigured realm. Use the example kafka-authz-realm.json file to import the realm. You can check the authorization rules defined for the realm in the Admin Console. The rules grant access to the resources on the Kafka cluster configured to use the example Red Hat Single Sign-On realm.

Prerequisites

A running OpenShift cluster.
The AMQ Streams examples/security/keycloak-authorization/kafka-authz-realm.json file that contains the preconfigured realm.

Procedure

Install the Red Hat Single Sign-On server using the Red Hat Single Sign-On Operator as described in Server Installation and Configuration in the Red Hat Single Sign-On documentation.
Wait until the Red Hat Single Sign-On instance is running.
Get the external hostname to be able to access the Admin Console.
```
NS=sso
oc get ingress keycloak -n $NS
```
In this example, we assume the Red Hat Single Sign-On server is running in the sso namespace.
Get the password for the admin user.
```
oc get -n $NS pod keycloak-0 -o yaml | less
```
The password is stored as a secret, so get the configuration YAML file for the Red Hat Single Sign-On instance to identify the name of the secret (secretKeyRef.name).
Use the name of the secret to obtain the clear text password.
```
SECRET_NAME=credential-keycloak
oc get -n $NS secret $SECRET_NAME -o yaml | grep PASSWORD | awk '{print $2}' | base64 -D
```
In this example, we assume the name of the secret is credential-keycloak.
Log in to the Admin Console with the username admin and the password you obtained.
Use https://HOSTNAME to access the OpenShift ingress.
You can now upload the example realm to Red Hat Single Sign-On using the Admin Console.
Click Add Realm to import the example realm.
Add the examples/security/keycloak-authorization/kafka-authz-realm.json file, and then click Create.
You now have kafka-authz as your current realm in the Admin Console.
The default view displays the Master realm.
In the Red Hat Single Sign-On Admin Console, go to Clients > kafka > Authorization > Settings and check that Decision Strategy is set to Affirmative.
An affirmative policy means that at least one policy must be satisfied for a client to access the Kafka cluster.
In the Red Hat Single Sign-On Admin Console, go to Groups, Users, Roles and Clients to view the realm configuration.
Groups
Groups are used to create user groups and set user permissions. Groups are sets of users with a name assigned. They are used to compartmentalize users into geographical, organizational or departmental units. Groups can be linked to an LDAP identity provider. You can make a user a member of a group through a custom LDAP server admin user interface, for example, to grant permissions on Kafka resources.
Users
Users are used to create users. For this example, alice and bob are defined. alice is a member of the ClusterManager group and bob is a member of ClusterManager-my-cluster group. Users can be stored in an LDAP identity provider.
Roles
Roles mark users or clients as having certain permissions. Roles are a concept analogous to groups. They are usually used to tag users with organizational roles and have the requisite permissions. Roles cannot be stored in an LDAP identity provider. If LDAP is a requirement, you can use groups instead, and add Red Hat Single Sign-On roles to the groups so that when users are assigned a group they also get a corresponding role.
Clients
Clients can have specific configurations. For this example, kafka, kafka-cli, team-a-client, and team-b-client clients are configured.
The kafka client is used by Kafka brokers to perform the necessary OAuth 2.0 communication for access token validation. This client also contains the authorization services resource definitions, policies, and authorization scopes used to perform authorization on the Kafka brokers. The authorization configuration is defined in the kafka client from the Authorization tab, which becomes visible when Authorization Enabled is switched on from the Settings tab.
The kafka-cli client is a public client that is used by the Kafka command line tools when authenticating with username and password to obtain an access token or a refresh token.
The team-a-client and team-b-client clients are confidential clients representing services with partial access to certain Kafka topics.
In the Red Hat Single Sign-On Admin Console, go to Authorization > Permissions to see the granted permissions that use the resources and policies defined for the realm.
For example, the kafka client has the following permissions:
```
Dev Team A can write to topics that start with x_ on any cluster
Dev Team B can read from topics that start with x_ on any cluster
Dev Team B can update consumer group offsets that start with x_ on any cluster
ClusterManager of my-cluster Group has full access to cluster config on my-cluster
ClusterManager of my-cluster Group has full access to consumer groups on my-cluster
ClusterManager of my-cluster Group has full access to topics on my-cluster
```
Dev Team A
The Dev Team A realm role can write to topics that start with x_ on any cluster. This combines a resource called Topic:x_*, Describe and Write scopes, and the Dev Team A policy. The Dev Team A policy matches all users that have a realm role called Dev Team A.
Dev Team B
The Dev Team B realm role can read from topics that start with x_ on any cluster. This combines Topic:x_*, Group:x_* resources, Describe and Read scopes, and the Dev Team B policy. The Dev Team B policy matches all users that have a realm role called Dev Team B. Matching users and clients have the ability to read from topics, and update the consumed offsets for topics and consumer groups that have names starting with x_.

6.5.4.2. Deploying a Kafka cluster with Red Hat Single Sign-On authorization

Deploy a Kafka cluster configured to connect to the Red Hat Single Sign-On server. Use the example kafka-ephemeral-oauth-single-keycloak-authz.yaml file to deploy the Kafka cluster as a Kafka custom resource. The example deploys a single-node Kafka cluster with keycloak authorization and oauth authentication.

Prerequisites

The Red Hat Single Sign-On authorization server is deployed to your OpenShift cluster and loaded with the example realm.
The Cluster Operator is deployed to your OpenShift cluster.
The AMQ Streams examples/security/keycloak-authorization/kafka-ephemeral-oauth-single-keycloak-authz.yaml custom resource.

Procedure

Use the hostname of the Red Hat Single Sign-On instance you deployed to prepare a truststore certificate for Kafka brokers to communicate with the Red Hat Single Sign-On server.
```
SSO_HOST=SSO-HOSTNAME
SSO_HOST_PORT=$SSO_HOST:443
STOREPASS=storepass

echo "Q" | openssl s_client -showcerts -connect $SSO_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/sso.crt
```
The certificate is required as OpenShift ingress is used to make a secure (HTTPS) connection.

Deploy the certificate to OpenShift as a secret.

oc create secret generic oauth-server-cert --from-file=/tmp/sso.crt -n $NS

Set the hostname as an environment variable
```
SSO_HOST=SSO-HOSTNAME
```

Create and deploy the example Kafka cluster.

cat examples/security/keycloak-authorization/kafka-ephemeral-oauth-single-keycloak-authz.yaml | sed -E 's#\${SSO_HOST}'"#$SSO_HOST#" | oc create -n $NS -f -

6.5.4.3. Preparing TLS connectivity for a CLI Kafka client session

Create a new pod for an interactive CLI session. Set up a truststore with a Red Hat Single Sign-On certificate for TLS connectivity. The truststore is to connect to Red Hat Single Sign-On and the Kafka broker.

Prerequisites

The Red Hat Single Sign-On authorization server is deployed to your OpenShift cluster and loaded with the example realm.
In the Red Hat Single Sign-On Admin Console, check the roles assigned to the clients are displayed in Clients > Service Account Roles.
The Kafka cluster configured to connect with Red Hat Single Sign-On is deployed to your OpenShift cluster.

Procedure

Run a new interactive pod container using the AMQ Streams Kafka image to connect to a running Kafka broker.
```
NS=sso
oc run -ti --restart=Never --image=registry.redhat.io/amq7/amq-streams-kafka-33-rhel8:2.3.0 kafka-cli -n $NS -- /bin/sh
```
Note
If oc times out waiting on the image download, subsequent attempts may result in an AlreadyExists error.
Attach to the pod container.
```
oc attach -ti kafka-cli -n $NS
```

Use the hostname of the Red Hat Single Sign-On instance to prepare a certificate for client connection using TLS.

SSO_HOST=SSO-HOSTNAME
SSO_HOST_PORT=$SSO_HOST:443
STOREPASS=storepass

echo "Q" | openssl s_client -showcerts -connect $SSO_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/sso.crt

Create a truststore for TLS connection to the Kafka brokers.

keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias sso -storepass $STOREPASS -import -file /tmp/sso.crt -noprompt

Use the Kafka bootstrap address as the hostname of the Kafka broker and the tls listener port (9093) to prepare a certificate for the Kafka broker.

KAFKA_HOST_PORT=my-cluster-kafka-bootstrap:9093
STOREPASS=storepass

echo "Q" | openssl s_client -showcerts -connect $KAFKA_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/my-cluster-kafka.crt

Add the certificate for the Kafka broker to the truststore.

keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias my-cluster-kafka -storepass $STOREPASS -import -file /tmp/my-cluster-kafka.crt -noprompt

Keep the session open to check authorized access.

6.5.4.4. Checking authorized access to Kafka using a CLI Kafka client session

Check the authorization rules applied through the Red Hat Single Sign-On realm using an interactive CLI session. Apply the checks using Kafka’s example producer and consumer clients to create topics with user and service accounts that have different levels of access.

Use the team-a-client and team-b-client clients to check the authorization rules. Use the alice admin user to perform additional administrative tasks on Kafka.

The AMQ Streams Kafka image used in this example contains Kafka producer and consumer binaries.

Prerequisites

ZooKeeper and Kafka are running in the OpenShift cluster to be able to send and receive messages.
The interactive CLI Kafka client session is started.
Apache Kafka download.

Setting up client and admin user configuration

Prepare a Kafka configuration file with authentication properties for the team-a-client client.

SSO_HOST=SSO-HOSTNAME

cat > /tmp/team-a-client.properties << EOF
security.protocol=SASL_SSL
ssl.truststore.location=/tmp/truststore.p12
ssl.truststore.password=$STOREPASS
ssl.truststore.type=PKCS12
sasl.mechanism=OAUTHBEARER
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.client.id="team-a-client" \
  oauth.client.secret="team-a-client-secret" \
  oauth.ssl.truststore.location="/tmp/truststore.p12" \
  oauth.ssl.truststore.password="$STOREPASS" \
  oauth.ssl.truststore.type="PKCS12" \
  oauth.token.endpoint.uri="https://$SSO_HOST/auth/realms/kafka-authz/protocol/openid-connect/token" ;
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
EOF

The SASL OAUTHBEARER mechanism is used. This mechanism requires a client ID and client secret, which means the client first connects to the Red Hat Single Sign-On server to obtain an access token. The client then connects to the Kafka broker and uses the access token to authenticate.

Prepare a Kafka configuration file with authentication properties for the team-b-client client.

cat > /tmp/team-b-client.properties << EOF
security.protocol=SASL_SSL
ssl.truststore.location=/tmp/truststore.p12
ssl.truststore.password=$STOREPASS
ssl.truststore.type=PKCS12
sasl.mechanism=OAUTHBEARER
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.client.id="team-b-client" \
  oauth.client.secret="team-b-client-secret" \
  oauth.ssl.truststore.location="/tmp/truststore.p12" \
  oauth.ssl.truststore.password="$STOREPASS" \
  oauth.ssl.truststore.type="PKCS12" \
  oauth.token.endpoint.uri="https://$SSO_HOST/auth/realms/kafka-authz/protocol/openid-connect/token" ;
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
EOF

Authenticate admin user alice by using curl and performing a password grant authentication to obtain a refresh token.

USERNAME=alice
PASSWORD=alice-password

GRANT_RESPONSE=$(curl -X POST "https://$SSO_HOST/auth/realms/kafka-authz/protocol/openid-connect/token" -H 'Content-Type: application/x-www-form-urlencoded' -d "grant_type=password&username=$USERNAME&password=$PASSWORD&client_id=kafka-cli&scope=offline_access" -s -k)

REFRESH_TOKEN=$(echo $GRANT_RESPONSE | awk -F "refresh_token\":\"" '{printf $2}' | awk -F "\"" '{printf $1}')

The refresh token is an offline token that is long-lived and does not expire.

Prepare a Kafka configuration file with authentication properties for the admin user alice.

cat > /tmp/alice.properties << EOF
security.protocol=SASL_SSL
ssl.truststore.location=/tmp/truststore.p12
ssl.truststore.password=$STOREPASS
ssl.truststore.type=PKCS12
sasl.mechanism=OAUTHBEARER
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.refresh.token="$REFRESH_TOKEN" \
  oauth.client.id="kafka-cli" \
  oauth.ssl.truststore.location="/tmp/truststore.p12" \
  oauth.ssl.truststore.password="$STOREPASS" \
  oauth.ssl.truststore.type="PKCS12" \
  oauth.token.endpoint.uri="https://$SSO_HOST/auth/realms/kafka-authz/protocol/openid-connect/token" ;
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
EOF

The kafka-cli public client is used for the oauth.client.id in the sasl.jaas.config. Since it’s a public client it does not require a secret. The client authenticates with the refresh token that was authenticated in the previous step. The refresh token requests an access token behind the scenes, which is then sent to the Kafka broker for authentication.

Producing messages with authorized access

Use the team-a-client configuration to check that you can produce messages to topics that start with a_ or x_.

Write to topic my-topic.
```
bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic my-topic \
  --producer.config=/tmp/team-a-client.properties
First message
```
This request returns a Not authorized to access topics: [my-topic] error.
team-a-client has a Dev Team A role that gives it permission to perform any supported actions on topics that start with a_, but can only write to topics that start with x_. The topic named my-topic matches neither of those rules.

Write to topic a_messages.

bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
  --producer.config /tmp/team-a-client.properties
First message
Second message

Messages are produced to Kafka successfully.

Press CTRL+C to exit the CLI application.
Check the Kafka container log for a debug log of Authorization GRANTED for the request.
```
oc logs my-cluster-kafka-0 -f -n $NS
```

Consuming messages with authorized access

Use the team-a-client configuration to consume messages from topic a_messages.

Fetch messages from topic a_messages.
```
bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
  --from-beginning --consumer.config /tmp/team-a-client.properties
```
The request returns an error because the Dev Team A role for team-a-client only has access to consumer groups that have names starting with a_.

Update the team-a-client properties to specify the custom consumer group it is permitted to use.

bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
  --from-beginning --consumer.config /tmp/team-a-client.properties --group a_consumer_group_1

The consumer receives all the messages from the a_messages topic.

Administering Kafka with authorized access

The team-a-client is an account without any cluster-level access, but it can be used with some administrative operations.

List topics.

bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list

The a_messages topic is returned.

List consumer groups.

bin/kafka-consumer-groups.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list

The a_consumer_group_1 consumer group is returned.

Fetch details on the cluster configuration.

bin/kafka-configs.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties \
  --entity-type brokers --describe --entity-default

The request returns an error because the operation requires cluster level permissions that team-a-client does not have.

Using clients with different permissions

Use the team-b-client configuration to produce messages to topics that start with b_.

Write to topic a_messages.

bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
  --producer.config /tmp/team-b-client.properties
Message 1

This request returns a Not authorized to access topics: [a_messages] error.

Write to topic b_messages.

bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic b_messages \
  --producer.config /tmp/team-b-client.properties
Message 1
Message 2
Message 3

Messages are produced to Kafka successfully.

Write to topic x_messages.

bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --producer.config /tmp/team-b-client.properties
Message 1

A Not authorized to access topics: [x_messages] error is returned, The team-b-client can only read from topic x_messages.

Write to topic x_messages using team-a-client.
```
bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --producer.config /tmp/team-a-client.properties
Message 1
```
This request returns a Not authorized to access topics: [x_messages] error. The team-a-client can write to the x_messages topic, but it does not have a permission to create a topic if it does not yet exist. Before team-a-client can write to the x_messages topic, an admin power user must create it with the correct configuration, such as the number of partitions and replicas.

Managing Kafka with an authorized admin user

Use admin user alice to manage Kafka. alice has full access to manage everything on any Kafka cluster.

Create the x_messages topic as alice.

bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/alice.properties \
  --topic x_messages --create --replication-factor 1 --partitions 1

The topic is created successfully.

List all topics as alice.

bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/alice.properties --list
bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list
bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-b-client.properties --list

Admin user alice can list all the topics, whereas team-a-client and team-b-client can only list the topics they have access to.

The Dev Team A and Dev Team B roles both have Describe permission on topics that start with x_, but they cannot see the other team’s topics because they do not have Describe permissions on them.

Use the team-a-client to produce messages to the x_messages topic:

bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --producer.config /tmp/team-a-client.properties
Message 1
Message 2
Message 3

As alice created the x_messages topic, messages are produced to Kafka successfully.

Use the team-b-client to produce messages to the x_messages topic.

bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --producer.config /tmp/team-b-client.properties
Message 4
Message 5

This request returns a Not authorized to access topics: [x_messages] error.

Use the team-b-client to consume messages from the x_messages topic:

bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --from-beginning --consumer.config /tmp/team-b-client.properties --group x_con