Chapter 9. Configuring a deployment

Procedure

1. Add the plugin properties to the config of the Kafka resource.

   The plugin properties are shown in this example configuration.

   Example Kafka Static Quota plugin configuration

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

   1

   Loads the Kafka Static Quota plugin.

   2

   Sets the producer byte-rate threshold. 1 MBps in this example.

   3

   Sets the consumer byte-rate threshold. 1 MBps in this example.

   4

   Sets the lower soft limit for storage. 400 GB in this example.

   5

   Sets the higher hard limit for storage. 500 GB in this example.

   6

   Sets the interval in seconds between checks on storage. 5 seconds in this example. You can set this to 0 to disable the check.

2. Update the resource.

   oc apply -f <kafka_configuration_file>

Procedure

1. Find the name of the Pod that you want to delete.

   Kafka broker pods are named <cluster_name>-kafka-<index_number>, where <index_number> starts at zero and ends at the total number of replicas minus one. For example, my-cluster-kafka-0.

2. Use oc annotate to annotate the Pod resource in OpenShift:

   oc annotate pod <cluster_name>-kafka-<index_number> strimzi.io/delete-pod-and-pvc="true"

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

Procedure

1. Find the name of the Pod that you want to delete.

   ZooKeeper pods are named <cluster_name>-zookeeper-<index_number>, where <index_number> starts at zero and ends at the total number of replicas minus one. For example, my-cluster-zookeeper-0.

2. Use oc annotate to annotate the Pod resource in OpenShift:

   oc annotate pod <cluster_name>-zookeeper-<index_number> strimzi.io/delete-pod-and-pvc="true"

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

Procedure

1. Annotate the node pool with the IDs to use when scaling up or scaling down, as shown in the following examples.

   IDs for scaling up are assigned to node pool pool-a:

   Assigning IDs for scaling up

   oc annotate kafkanodepool pool-a strimzi.io/next-node-ids="[0,1,2,10-20,30]"

   The lowest available ID from this range is used when adding a node to pool-a.

   IDs for scaling down are assigned to node pool pool-b:

   Assigning IDs for scaling down

   oc annotate kafkanodepool pool-b strimzi.io/remove-node-ids="[60-50,9,8,7]"

   The highest available ID from this range is removed when scaling down pool-b.

   Note

   If you want to remove a specific node, you can assign a single node ID to the scaling down annotation: oc annotate kafkanodepool pool-b strimzi.io/remove-node-ids="[3]".

2. You can now scale the node pool.

   For more information, see the following:

   • Section 9.3.3, “Adding nodes to a node pool”
   • Section 9.3.4, “Removing nodes from a node pool”
   • Section 9.3.5, “Moving nodes between node pools”

   On reconciliation, a warning is given if the annotations are misformatted.

3. After you have performed the scaling operation, you can remove the annotation if it’s no longer needed.

   Removing the annotation for scaling up

   oc annotate kafkanodepool pool-a strimzi.io/next-node-ids-

   Removing the annotation for scaling down

   oc annotate kafkanodepool pool-b strimzi.io/remove-node-ids-

Procedure

1. Create a new node in the node pool.

   For example, node pool pool-a has three replicas. We add a node by increasing the number of replicas:

   oc scale kafkanodepool pool-a --replicas=4

2. Check the status of the deployment and wait for the pods in the node pool to be created and ready (1/1).

   oc get pods -n <my_cluster_operator_namespace>

   Output shows four Kafka nodes in the node pool

   NAME                 READY  STATUS   RESTARTS
   my-cluster-pool-a-0  1/1    Running  0
   my-cluster-pool-a-1  1/1    Running  0
   my-cluster-pool-a-2  1/1    Running  0
   my-cluster-pool-a-3  1/1    Running  0

3. Reassign the partitions after increasing the number of nodes in the node pool.

   After scaling up a node pool, use the Cruise Control add-brokers mode to move partition replicas from existing brokers to the newly added brokers.

   Using Cruise Control to reassign partition replicas

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaRebalance
   metadata:
     # ...
   spec:
     mode: add-brokers
     brokers: [3]

   We are reassigning partitions to node my-cluster-pool-a-3. The reassignment can take some time depending on the number of topics and partitions in the cluster.

Procedure

1. Reassign the partitions before decreasing the number of nodes in the node pool.

   Before scaling down a node pool, use the Cruise Control remove-brokers mode to move partition replicas off the brokers that are going to be removed.

   Using Cruise Control to reassign partition replicas

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaRebalance
   metadata:
     # ...
   spec:
     mode: remove-brokers
     brokers: [3]

   We are reassigning partitions from node my-cluster-pool-a-3. The reassignment can take some time depending on the number of topics and partitions in the cluster.

2. After the reassignment process is complete, and the node being removed has no live partitions, reduce the number of Kafka nodes in the node pool.

   For example, node pool pool-a has four replicas. We remove a node by decreasing the number of replicas:

   oc scale kafkanodepool pool-a --replicas=3

   Output shows three Kafka nodes in the node pool

   NAME                       READY  STATUS   RESTARTS
   my-cluster-pool-b-kafka-0  1/1    Running  0
   my-cluster-pool-b-kafka-1  1/1    Running  0
   my-cluster-pool-b-kafka-2  1/1    Running  0

Procedure

1. Create a new node in the target node pool.

   For example, node pool pool-a has three replicas. We add a node by increasing the number of replicas:

   oc scale kafkanodepool pool-a --replicas=4

2. Check the status of the deployment and wait for the pods in the node pool to be created and ready (1/1).

   oc get pods -n <my_cluster_operator_namespace>

   Output shows four Kafka nodes in the source and target node pools

   NAME                 READY  STATUS   RESTARTS
   my-cluster-pool-a-0  1/1    Running  0
   my-cluster-pool-a-1  1/1    Running  0
   my-cluster-pool-a-4  1/1    Running  0
   my-cluster-pool-a-7  1/1    Running  0
   my-cluster-pool-b-2  1/1    Running  0
   my-cluster-pool-b-3  1/1    Running  0
   my-cluster-pool-b-5  1/1    Running  0
   my-cluster-pool-b-6  1/1    Running  0

   Node IDs are appended to the name of the node on creation. We add node my-cluster-pool-a-7, which has a node ID of 7.

3. Reassign the partitions from the old node to the new node.

   Before scaling down the source node pool, use the Cruise Control remove-brokers mode to move partition replicas off the brokers that are going to be removed.

   Using Cruise Control to reassign partition replicas

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaRebalance
   metadata:
     # ...
   spec:
     mode: remove-brokers
     brokers: [6]

   We are reassigning partitions from node my-cluster-pool-b-6. The reassignment can take some time depending on the number of topics and partitions in the cluster.

4. After the reassignment process is complete, reduce the number of Kafka nodes in the source node pool.

   For example, node pool pool-b has four replicas. We remove a node by decreasing the number of replicas:

   oc scale kafkanodepool pool-b --replicas=3

   The node with the highest ID (6) within the pool is removed.

   Output shows three Kafka nodes in the source node pool

   NAME                       READY  STATUS   RESTARTS
   my-cluster-pool-b-kafka-2  1/1    Running  0
   my-cluster-pool-b-kafka-3  1/1    Running  0
   my-cluster-pool-b-kafka-5  1/1    Running  0

Procedure

1. Create a node pool with a broker role.

   Example node pool configuration

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

   The new node pool also has three nodes. If you already have a broker-only node pool, you can skip this step.

2. Apply the new KafkaNodePool resource to create the brokers.

3. Check the status of the deployment and wait for the pods in the node pool to be created and ready (1/1).

   oc get pods -n <my_cluster_operator_namespace>

   Output shows pods running in two node pools

   NAME                 READY  STATUS   RESTARTS
   my-cluster-pool-a-0  1/1    Running  0
   my-cluster-pool-a-1  1/1    Running  0
   my-cluster-pool-a-2  1/1    Running  0
   my-cluster-pool-b-3  1/1    Running  0
   my-cluster-pool-b-4  1/1    Running  0
   my-cluster-pool-b-5  1/1    Running  0

   Node IDs are appended to the name of the node on creation.

4. Use the Cruise Control remove-brokers mode to reassign partition replicas from the dual-role nodes to the newly added brokers.

   Using Cruise Control to reassign partition replicas

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaRebalance
   metadata:
     # ...
   spec:
     mode: remove-brokers
     brokers: [0, 1, 2]

   The reassignment can take some time depending on the number of topics and partitions in the cluster.

   Note

   If nodes changing to controller-only roles have any assigned partitions, the change is prevented. The status.conditions of the Kafka resource provide details of events preventing the change.

5. Remove the broker role from the node pool that originally had a combined role.

   Dual-role nodes switched to controllers

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

6. Apply the configuration change so that the node pool switches to a controller-only role.

Procedure

1. Edit the node pool pool-a and add the broker role to it.

   Example node pool configuration

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

2. Check the status and wait for the pods in the node pool to be restarted and ready (1/1).

   oc get pods -n <my_cluster_operator_namespace>

   Output shows pods running in two node pools

   NAME                 READY  STATUS   RESTARTS
   my-cluster-pool-a-0  1/1    Running  0
   my-cluster-pool-a-1  1/1    Running  0
   my-cluster-pool-a-2  1/1    Running  0
   my-cluster-pool-b-3  1/1    Running  0
   my-cluster-pool-b-4  1/1    Running  0
   my-cluster-pool-b-5  1/1    Running  0

   Node IDs are appended to the name of the node on creation.

3. Use the Cruise Control remove-brokers mode to reassign partition replicas from the broker-only nodes to the dual-role nodes.

   Using Cruise Control to reassign partition replicas

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaRebalance
   metadata:
     # ...
   spec:
     mode: remove-brokers
     brokers: [3, 4, 5]

   The reassignment can take some time depending on the number of topics and partitions in the cluster.

4. Remove the pool-b node pool that has the old broker-only nodes.

   oc delete kafkanodepool pool-b -n <my_cluster_operator_namespace>

Procedure

1. Create the node pool with its own storage settings.

   For example, node pool pool-a uses JBOD storage with persistent volumes:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaNodePool
   metadata:
     name: pool-a
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     replicas: 3
     storage:
       type: jbod
       volumes:
         - id: 0
           type: persistent-claim
           size: 500Gi
           class: gp2-ebs
     # ...

   Nodes in pool-a are configured to use Amazon EBS (Elastic Block Store) GP2 volumes.

2. Apply the node pool configuration for pool-a.

3. Check the status of the deployment and wait for the pods in pool-a to be created and ready (1/1).

   oc get pods -n <my_cluster_operator_namespace>

   Output shows three Kafka nodes in the node pool

   NAME                 READY  STATUS   RESTARTS
   my-cluster-pool-a-0  1/1    Running  0
   my-cluster-pool-a-1  1/1    Running  0
   my-cluster-pool-a-2  1/1    Running  0

4. To migrate to a new storage class, create a new node pool with the required storage configuration:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaNodePool
   metadata:
     name: pool-b
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     roles:
       - broker
     replicas: 3
     storage:
       type: jbod
       volumes:
         - id: 0
           type: persistent-claim
           size: 1Ti
           class: gp3-ebs
     # ...

   Nodes in pool-b are configured to use Amazon EBS (Elastic Block Store) GP3 volumes.

5. Apply the node pool configuration for pool-b.
6. Check the status of the deployment and wait for the pods in pool-b to be created and ready.

7. Reassign the partitions from pool-a to pool-b.

   When migrating to a new storage configuration, use the Cruise Control remove-brokers mode to move partition replicas off the brokers that are going to be removed.

   Using Cruise Control to reassign partition replicas

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaRebalance
   metadata:
     # ...
   spec:
     mode: remove-brokers
     brokers: [0, 1, 2]

   We are reassigning partitions from pool-a. The reassignment can take some time depending on the number of topics and partitions in the cluster.

8. After the reassignment process is complete, delete the old node pool:

   oc delete kafkanodepool pool-a

Procedure

1. Define the storage class for use with each availability zone:

   apiVersion: storage.k8s.io/v1
   kind: StorageClass
   metadata:
     name: all-zones
   provisioner: kubernetes.io/my-storage
   parameters:
     type: ssd
   volumeBindingMode: WaitForFirstConsumer

2. Create node pools representing the two availability zones, specifying the all-zones storage class and the affinity for each zone:

   Node pool configuration for zone-1

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaNodePool
   metadata:
     name: pool-zone-1
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     replicas: 3
     storage:
       type: jbod
       volumes:
         - id: 0
           type: persistent-claim
           size: 500Gi
           class: all-zones
     template:
       pod:
         affinity:
           nodeAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               nodeSelectorTerms:
                 - matchExpressions:
                   - key: topology.kubernetes.io/zone
                     operator: In
                     values:
                     - zone-1
     # ...

   Node pool configuration for zone-2

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaNodePool
   metadata:
     name: pool-zone-2
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     replicas: 4
     storage:
       type: jbod
       volumes:
         - id: 0
           type: persistent-claim
           size: 500Gi
           class: all-zones
     template:
       pod:
         affinity:
           nodeAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               nodeSelectorTerms:
                 - matchExpressions:
                   - key: topology.kubernetes.io/zone
                     operator: In
                     values:
                     - zone-2
     # ...

3. Apply the node pool configuration.

4. Check the status of the deployment and wait for the pods in the node pools to be created and ready (1/1).

   oc get pods -n <my_cluster_operator_namespace>

   Output shows 3 Kafka nodes in pool-zone-1 and 4 Kafka nodes in pool-zone-2

   NAME                            READY  STATUS   RESTARTS
   my-cluster-pool-zone-1-kafka-0  1/1    Running  0
   my-cluster-pool-zone-1-kafka-1  1/1    Running  0
   my-cluster-pool-zone-1-kafka-2  1/1    Running  0
   my-cluster-pool-zone-2-kafka-3  1/1    Running  0
   my-cluster-pool-zone-2-kafka-4  1/1    Running  0
   my-cluster-pool-zone-2-kafka-5  1/1    Running  0
   my-cluster-pool-zone-2-kafka-6  1/1    Running  0

Procedure

1. Create a new KafkaNodePool resource.

   1. Name the resource kafka.
   2. Point a strimzi.io/cluster label to your existing Kafka resource.
   3. Set the replica count and storage configuration to match your current Kafka cluster.
   4. Set the roles to broker.

   Example configuration for a node pool used in migrating a Kafka cluster

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

   Warning

   To preserve cluster data and the names of its nodes and resources, the node pool name must be kafka, and the strimzi.io/cluster label must match the Kafka resource name. Otherwise, nodes and resources are created with new names, including the persistent volume storage used by the nodes. Consequently, your previous data may not be available.

2. Apply the KafkaNodePool resource:

   oc apply -f <node_pool_configuration_file>

   By applying this resource, you switch Kafka to using node pools.

   There is no change or rolling update and resources are identical to how they were before.

3. Enable support for node pools in the Kafka resource using the strimzi.io/node-pools: enabled annotation.

   Example configuration for a node pool in a cluster using ZooKeeper

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
     annotations:
       strimzi.io/node-pools: enabled
   spec:
     kafka:
       # ...
     zookeeper:
       # ...

4. Apply the Kafka resource:

   oc apply -f <kafka_configuration_file>

   There is no change or rolling update. The resources remain identical to how they were before.

5. Remove the replicated properties from the Kafka custom resource. When the KafkaNodePool resource is in use, you can remove the properties that you copied to the KafkaNodePool resource, such as the .spec.kafka.replicas and .spec.kafka.storage properties.

1. If you have multiple node pools, consolidate them into a single KafkaNodePool named kafka with node IDs from 0 to N (where N is the number of replicas).
2. Ensure that the .spec.kafka configuration in the Kafka resource matches the KafkaNodePool configuration, including storage, resources, and replicas.
3. Disable support for node pools in the Kafka resource using the strimzi.io/node-pools: disabled annotation.
4. Delete the Kafka node pool named kafka.

Procedure

1. Annotate the custom resource in OpenShift, setting pause-reconciliation to true:

   oc annotate <kind_of_custom_resource> <name_of_custom_resource> strimzi.io/pause-reconciliation="true"

   For example, for the KafkaConnect custom resource:

   oc annotate KafkaConnect my-connect strimzi.io/pause-reconciliation="true"

2. Check that the status conditions of the custom resource show a change to ReconciliationPaused:

   oc describe <kind_of_custom_resource> <name_of_custom_resource>

   The type condition changes to ReconciliationPaused at the lastTransitionTime.

   Example custom resource with a paused reconciliation condition type

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

Procedure

1. To add proxy environment variables to the Cluster Operator, update its Deployment configuration (install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml).

   Example proxy configuration for the Cluster Operator

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         serviceAccountName: strimzi-cluster-operator
         containers:
           # ...
           env:
           # ...
           - name: "HTTP_PROXY"
             value: "http://proxy.com" 1
           - name: "HTTPS_PROXY"
             value: "https://proxy.com" 2
           - name: "NO_PROXY"
             value: "internal.com, other.domain.com" 3
     # ...

   1

   Address of the proxy server.

   2

   Secure address of the proxy server.

   3

   Addresses for servers that are accessed directly as exceptions to the proxy server. The URLs are comma-separated.

   Alternatively, edit the Deployment directly:

   oc edit deployment strimzi-cluster-operator

2. If you updated the YAML file instead of editing the Deployment directly, apply the changes:

   oc create -f install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml

Procedure

1. To disable the FIPS mode in the Cluster Operator, update its Deployment configuration (install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml) and add the FIPS_MODE environment variable.

   Example FIPS configuration for the Cluster Operator

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         serviceAccountName: strimzi-cluster-operator
         containers:
           # ...
           env:
           # ...
           - name: "FIPS_MODE"
             value: "disabled" 1
     # ...

   1

   Disables the FIPS mode.

   Alternatively, edit the Deployment directly:

   oc edit deployment strimzi-cluster-operator

2. If you updated the YAML file instead of editing the Deployment directly, apply the changes:

   oc apply -f install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml

Procedure

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

   Access rights are configured for the Kafka Connect topics and cluster group using literal name values. The following table shows the default names configured for the topics and cluster group ID.

   Table 9.2. Names for the access rights configuration

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

   In this example configuration, the default names are used to specify access rights. If you are using different names for a Kafka Connect instance, use those names in the ACLs configuration.

   Example configuration for simple authorization

   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: "*"
         # cluster group
         - resource:
             type: group
             name: connect-cluster
             patternType: literal
           operations:
             - Read
           host: "*"

2. Create or update the resource.

   oc apply -f KAFKA-USER-CONFIG-FILE

Procedure

1. Find the name of the KafkaConnector custom resource that controls the connector you want to pause or stop:

   oc get KafkaConnector

2. Edit the KafkaConnector resource to stop or pause the connector.

   Example configuration for stopping a Kafka Connect connector

   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:
       file: "/opt/kafka/LICENSE"
       topic: my-topic
     state: stopped
     # ...

   Change the state configuration to stopped or paused. The default state for the connector when this property is not set is running.

3. Apply the changes to the KafkaConnector configuration.

   You can resume the connector by changing state to running or removing the configuration.

Procedure

1. Find the name of the KafkaConnector custom resource that controls the Kafka connector you want to restart:

   oc get KafkaConnector

2. Restart the connector by annotating the KafkaConnector resource in OpenShift:

   oc annotate KafkaConnector <kafka_connector_name> strimzi.io/restart="true"

   The restart annotation is set to true.

3. Wait for the next reconciliation to occur (every two minutes by default).

   The Kafka connector is restarted, as long as the annotation was detected by the reconciliation process. When Kafka Connect accepts the restart request, the annotation is removed from the KafkaConnector custom resource.

Procedure

1. Find the name of the KafkaConnector custom resource that controls the Kafka connector task you want to restart:

   oc get KafkaConnector

2. Find the ID of the task to be restarted from the KafkaConnector custom resource:

   oc describe KafkaConnector <kafka_connector_name>

   Task IDs are non-negative integers, starting from 0.

3. Use the ID to restart the connector task by annotating the KafkaConnector resource in OpenShift:

   oc annotate KafkaConnector <kafka_connector_name> strimzi.io/restart-task="0"

   In this example, task 0 is restarted.

4. Wait for the next reconciliation to occur (every two minutes by default).

   The Kafka connector task is restarted, as long as the annotation was detected by the reconciliation process. When Kafka Connect accepts the restart request, the annotation is removed from the KafkaConnector custom resource.

Procedure

1. 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.7.0
       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.7"
       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.7.0
       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.7"
       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: {}

2. 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.

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

   1. 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.
   2. Configure the ACLs needed by MirrorMaker 2 to allow operations on the source and target Kafka clusters.

   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:
         # cluster group
         - resource:
             type: group
             name: mirrormaker2-cluster
           operations:
             - Read
         # access to config.storage.topic
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         # access to status.storage.topic
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         # access to offset.storage.topic
         - 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 the KafkaUserSpec schema reference.

4. 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.

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

   Example MirrorMaker 2 configuration with TLS encryption and mTLS authentication

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker-2
   spec:
     version: 3.7.0
     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.

6. 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>

Procedure

1. Find the name of the KafkaMirrorMaker2 custom resource that controls the MirrorMaker 2 connector you want to pause or stop:

   oc get KafkaMirrorMaker2

2. Edit the KafkaMirrorMaker2 resource to stop or pause the connector.

   Example configuration for stopping a MirrorMaker 2 connector

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker2
   spec:
     version: 3.7.0
     replicas: 3
     connectCluster: "my-cluster-target"
     clusters:
       # ...
     mirrors:
     - sourceCluster: "my-cluster-source"
       targetCluster: "my-cluster-target"
       sourceConnector:
         tasksMax: 10
         autoRestart:
           enabled: true
         state: stopped
     # ...

   Change the state configuration to stopped or paused. The default state for the connector when this property is not set is running.

3. Apply the changes to the KafkaMirrorMaker2 configuration.

   You can resume the connector by changing state to running or removing the configuration.

Procedure

1. Find the name of the KafkaMirrorMaker2 custom resource that controls the Kafka MirrorMaker 2 connector you want to restart:

   oc get KafkaMirrorMaker2

2. Find the name of the Kafka MirrorMaker 2 connector to be restarted from the KafkaMirrorMaker2 custom resource:

   oc describe KafkaMirrorMaker2 <mirrormaker_cluster_name>

3. Use the name of the connector to restart the connector by annotating the KafkaMirrorMaker2 resource in OpenShift:

   oc annotate KafkaMirrorMaker2 <mirrormaker_cluster_name> "strimzi.io/restart-connector=<mirrormaker_connector_name>"

   In this example, connector my-connector in the my-mirror-maker-2 cluster is restarted:

   oc annotate KafkaMirrorMaker2 my-mirror-maker-2 "strimzi.io/restart-connector=my-connector"

4. Wait for the next reconciliation to occur (every two minutes by default).

   The MirrorMaker 2 connector is restarted, as long as the annotation was detected by the reconciliation process. When MirrorMaker 2 accepts the request, the annotation is removed from the KafkaMirrorMaker2 custom resource.

Procedure

1. Find the name of the KafkaMirrorMaker2 custom resource that controls the MirrorMaker 2 connector task you want to restart:

   oc get KafkaMirrorMaker2

2. Find the name of the connector and the ID of the task to be restarted from the KafkaMirrorMaker2 custom resource:

   oc describe KafkaMirrorMaker2 <mirrormaker_cluster_name>

   Task IDs are non-negative integers, starting from 0.

3. Use the name and ID to restart the connector task by annotating the KafkaMirrorMaker2 resource in OpenShift:

   oc annotate KafkaMirrorMaker2 <mirrormaker_cluster_name> "strimzi.io/restart-connector-task=<mirrormaker_connector_name>:<task_id>"

   In this example, task 0 for connector my-connector in the my-mirror-maker-2 cluster is restarted:

   oc annotate KafkaMirrorMaker2 my-mirror-maker-2 "strimzi.io/restart-connector-task=my-connector:0"

4. Wait for the next reconciliation to occur (every two minutes by default).

   The MirrorMaker 2 connector task is restarted, as long as the annotation was detected by the reconciliation process. When MirrorMaker 2 accepts the request, the annotation is removed from the KafkaMirrorMaker2 custom resource.

Procedure

1. 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:
       # ...

2. 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.

3. 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.

Procedure

1. 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:
       # ...

2. Create or update the resource:

   oc apply -f <kafka_configuration_file>

3. Create new topics or reassign existing partitions to the new disks.

   Tip

   Cruise Control is an effective tool for reassigning partitions. To perform an intra-broker disk balance, you set rebalanceDisk to true under the KafkaRebalance.spec.

Procedure

1. 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.

   Tip

   You can use the kafka-reassign-partitions.sh tool to reassign the partitions.

2. 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:
       # ...

3. Create or update the resource:

   oc apply -f <kafka_configuration_file>

Procedure

1. 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.

2. 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.

3. Create or update the resource.

   oc apply -f <kafka_configuration_file>

Procedure

1. 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:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f <kafka_configuration_file>

Procedure

1. Label the nodes where Streams for Apache Kafka 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.

2. 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:
       # ...

3. Create or update the resource.

   This can be done using oc apply:

   oc apply -f <kafka_configuration_file>

Procedure

1. Select the nodes which should be used as dedicated.
2. Make sure there are no workloads scheduled on these nodes.

3. 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

4. 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

5. 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:
       # ...

6. Create or update the resource.

   This can be done using oc apply:

   oc apply -f <kafka_configuration_file>

Procedure

1. 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"
   # ...

2. 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.

   # ...
   logging:
     type: external
     valueFrom:
       configMapKeyRef:
         name: logging-configmap
         key: log4j.properties
   # ...

3. Create or update the resource.

   oc apply -f <kafka_configuration_file>

Procedure

1. 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

2. 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

Procedure

1. 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)
   # ...

2. 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

3. Apply the changes by deploying the Cluster Operator:

Procedure

1. Configure the KafkaConnect resource.

   • Enable the KubernetesConfigMapConfigProvider

   The specification shown here can support loading values from config maps and secrets.

   Example Kafka Connect configuration to use config maps and secrets

   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.configmaps.class: io.strimzi.kafka.KubernetesConfigMapConfigProvider 2
       config.providers.secrets.class: io.strimzi.kafka.KubernetesSecretConfigProvider 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

   KubernetesConfigMapConfigProvider provides values from config maps.

   3

   KubernetesSecretConfigProvider provides values from secrets.

2. Create or update the resource to enable the provider.

   oc apply -f <kafka_connect_configuration_file>

3. 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.

4. 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.

5. 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}
       # ...
   # ...

   The placeholder structure is configmaps:<path_and_file_name>:<property>. KubernetesConfigMapConfigProvider reads and extracts the option1 property value from the external config map.

Procedure

1. Configure the KafkaConnect resource.

   • Enable the EnvVarConfigProvider
   • Specify the environment variables using the externalConfiguration property.

   Example Kafka Connect configuration to use external environment variables

   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: org.apache.kafka.common.config.provider.EnvVarConfigProvider 2
     # ...
     externalConfiguration:
       env:
         - name: AWS_ACCESS_KEY_ID 3
           valueFrom:
             secretKeyRef:
               name: aws-creds 4
               key: awsAccessKey 5
         - name: AWS_SECRET_ACCESS_KEY
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsSecretAccessKey
     # ...

   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 environment variable takes a value from the secret.

   4

   The name of the secret containing the environment variable.

   5

   The name of the key stored in the secret.

   Note

   The secretKeyRef property references keys in a secret. If you are using a config map instead of a secret, use the configMapKeyRef property.

2. Create or update the resource to enable the provider.

   oc apply -f <kafka_connect_configuration_file>

3. 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:AWS_ACCESS_KEY_ID}
       option: ${env:AWS_SECRET_ACCESS_KEY}
       # ...
   # ...

   The placeholder structure is env:<environment_variable_name>. EnvVarConfigProvider reads and extracts the environment variable values from the mounted secret.

Procedure

1. Configure the KafkaConnect resource.

   • Enable the FileConfigProvider
   • Specify the file using the externalConfiguration property.

   Example Kafka Connect configuration to use an external property file

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     config:
       config.providers: file 1
       config.providers.file.class: org.apache.kafka.common.config.provider.FileConfigProvider 2
     #...
     externalConfiguration:
       volumes:
         - name: connector-config 3
           secret:
             secretName: mysecret 4

   1

   The alias for the configuration provider is used to define other configuration parameters.

   2

   FileConfigProvider provides values from properties files. The parameter uses the alias from config.providers, taking the form config.providers.${alias}.class.

   3

   The name of the volume containing the secret.

   4

   The name of the secret.

2. Create or update the resource to enable the provider.

   oc apply -f <kafka_connect_configuration_file>

3. Reference the file properties in the connector configuration as placeholders.

   Example connector configuration referencing the file

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     class: io.debezium.connector.mysql.MySqlConnector
     tasksMax: 2
     config:
       database.hostname: 192.168.99.1
       database.port: "3306"
       database.user: "${file:/opt/kafka/external-configuration/connector-config/mysecret:dbUsername}"
       database.password: "${file:/opt/kafka/external-configuration/connector-config/mysecret:dbPassword}"
       database.server.id: "184054"
       #...

   The placeholder structure is file:<path_and_file_name>:<property>. FileConfigProvider reads and extracts the database username and password property values from the mounted secret.

Procedure

1. Configure the KafkaConnect resource.

   • Enable the DirectoryConfigProvider
   • Specify the files using the externalConfiguration property.

   Example Kafka Connect configuration to use external property files

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     config:
       config.providers: directory 1
       config.providers.directory.class: org.apache.kafka.common.config.provider.DirectoryConfigProvider 2
     #...
     externalConfiguration:
       volumes: 3
         - name: cluster-ca 4
           secret:
             secretName: my-cluster-cluster-ca-cert 5
         - name: my-user
           secret:
             secretName: my-user 6

   1

   The alias for the configuration provider is used to define other configuration parameters.

   2

   DirectoryConfigProvider provides values from files in a directory. The parameter uses the alias from config.providers, taking the form config.providers.${alias}.class.

   3

   The names of the volumes containing the secrets.

   4

   The name of the secret for the cluster CA certificate to supply truststore configuration.

   5

   The name of the secret for the user to supply keystore configuration.

2. Create or update the resource to enable the provider.

   oc apply -f <kafka_connect_configuration_file>

3. Reference the file properties in the connector configuration as placeholders.

   Example connector configuration referencing the files

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     class: io.debezium.connector.mysql.MySqlConnector
     tasksMax: 2
     config:
       # ...
       database.history.producer.security.protocol: SSL
       database.history.producer.ssl.truststore.type: PEM
       database.history.producer.ssl.truststore.certificates: "${directory:/opt/kafka/external-configuration/cluster-ca:ca.crt}"
       database.history.producer.ssl.keystore.type: PEM
       database.history.producer.ssl.keystore.certificate.chain: "${directory:/opt/kafka/external-configuration/my-user:user.crt}"
       database.history.producer.ssl.keystore.key: "${directory:/opt/kafka/external-configuration/my-user:user.key}"
       #...

   The placeholder structure is directory:<path>:<file_name>. DirectoryConfigProvider reads and extracts the credentials from the mounted secrets.