Chapter 10. Configuring a deployment

Procedure

1. Add the plugin configuration to the quotas section of the Kafka resource.

   Example strimzi plugin configuration

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       quotas:
         type: strimzi
         producerByteRate: 1000000 

   1

   
         consumerByteRate: 1000000 

   2

   
         minAvailableBytesPerVolume: 500000000000 

   3

   
         excludedPrincipals: 

   4

   
           - my-user

   1

   Sets a producer byte-rate threshold of 1 MBps.

   2

   Sets a consumer byte-rate threshold of 1 MBps.

   3

   Sets an available bytes limit for storage of 500 GB.

   4

   Excludes my-user from the restrictions.

   minAvailableBytesPerVolume and minAvailableRatioPerVolume are mutually exclusive. Only configure one of these parameters.

   Example kafka plugin configuration

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       quotas:
         type: kafka
         producerByteRate: 1000000
         consumerByteRate: 1000000
         requestPercentage: 55 

   1

   
         controllerMutationRate: 50 

   2

   1

   Sets the CPU utilization limit to 55%.

   2

   Sets the controller mutation rate to 50 operations per second.

2. Apply the changes to the Kafka configuration.

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. 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 10.3.3, “Adding nodes to a node pool”
   • Section 10.3.4, “Removing nodes from a node pool”
   • Section 10.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.

   • If auto-rebalancing is enabled, partitions are reassigned to new nodes automatically, so you can skip this step.

   • If auto-rebalancing is not enabled, 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.

   • If auto-rebalancing is enabled, partitions are moved off brokers that are going to be removed automatically, so you can skip this step.

   • If auto-rebalancing is not enabled, 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.

   If auto-rebalancing is enabled, partitions are reassigned to new nodes and moved off brokers that are going to be removed automatically, so you can skip the next step.

3. If auto-rebalancing is not enabled, reassign partitions before decreasing the number of nodes in 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:
     roles:
       - broker
     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. 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.

   Expand

   Table 10.1. 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

   Show more

   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.

   You can also refine the restart behavior with the includeTasks and onlyFailed parameters, which both default to false.

   • includeTasks restarts both the connector instance and its task instances.
   • onlyFailed restarts only instances with a FAILED status when set to true.

   For example:

   strimzi.io/restart="includeTasks,onlyFailed"

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. Edit the KafkaConnector resource for the connector to include the listOffsets configuration.

   Example configuration to list offsets

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     listOffsets:
       toConfigMap: 

   1

   
         name: my-connector-offsets 

   2

   
     # ...

   1

   The reference to the config map where the list of offsets will be written to.

   2

   The name of the config map, which is named my-connector-offsets in this example.

2. Run the command to write the list to the config map by annotating the KafkaConnector resource:

   oc annotate kafkaconnector my-source-connector strimzi.io/connector-offsets=list -n <namespace>

   The annotation remains until either the list operation succeeds or it is manually removed from the resource.

3. After the KafkaConnector resource is updated, use the following command to check if the config map with the offsets was created:

   oc get configmap my-connector-offsets -n <namespace>

4. Inspect the contents of the config map to verify the offsets are being listed:

   oc describe configmap my-connector-offsets -n <namespace>

   Streams for Apache Kafka puts the offset information into the offsets.json property. This does not overwrite any other properties when updating an existing config map.

   Example source connector offset list

   apiVersion: v1
   kind: ConfigMap
   metadata:
     # ...
     ownerReferences: 

   1

   
     - apiVersion: kafka.strimzi.io/v1beta2
       blockOwnerDeletion: false
       controller: false
       kind: KafkaConnector
       name: my-source-connector
       uid: 637e3be7-bd96-43ab-abde-c55b4c4550e0
     resourceVersion: "66951"
     uid: 641d60a9-36eb-4f29-9895-8f2c1eb9638e
   data:
     offsets.json: |-
       {
         "offsets" : [ {
           "partition" : {
             "filename" : "/data/myfile.txt" 

   2

   
           },
           "offset" : {
             "position" : 15295 

   3

   
           }
         } ]
       }

   1

   The owner reference pointing to the KafkaConnector resource for the source connector. To provide a custom owner reference, create the config map in advance and set the owner reference.

   2

   The source partition, represented by the filename /data/myfile.txt in this example for a file-based connector.

   3

   The last processed offset position in the source partition.

   Example sink connector offset list

   apiVersion: v1
   kind: ConfigMap
   metadata:
     # ...
     ownerReferences: 

   1

   
     - apiVersion: kafka.strimzi.io/v1beta2
       blockOwnerDeletion: false
       controller: false
       kind: KafkaConnector
       name: my-sink-connector
       uid: 84a29d7f-77e6-43ac-bfbb-719f9b9a4b3b
     resourceVersion: "79241"
     uid: 721e30bc-23df-41a2-9b48-fb2b7d9b042c
   data:
     offsets.json: |-
       {
         "offsets": [
           {
             "partition": {
               "kafka_topic": "my-topic", 

   2

   
               "kafka_partition": 2 

   3

   
             },
             "offset": {
               "kafka_offset": 4 

   4

   
             }
           }
         ]
       }

   1

   The owner reference pointing to the KafkaConnector resource for the sink connector.

   2

   The Kafka topic that the sink connector is consuming from.

   3

   The partition of the Kafka topic.

   4

   The last committed Kafka offset for this topic and partition.

Procedure

1. Edit the KafkaConnector resource to stop the connector and include the alterOffsets configuration.

   Example configuration to stop a connector and alter offsets

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     state: stopped 

   1

   
     alterOffsets:
       fromConfigMap: 

   2

   
         name: my-connector-offsets 

   3

   
     # ...

   1

   Changes the state of the connector to stopped. The default state for the connector when this property is not set is running.

   2

   The reference to the config map that provides the update.

   3

   The name of the config map, which is named my-connector-offsets in this example.

2. Edit the config map to make the alteration.

   In this example, we’re resetting the offset position for a source connector to 15000.

   Example source connector offset list configuration

   apiVersion: v1
   kind: ConfigMap
   metadata:
     # ...
   data:
     offsets.json: |- 

   1

   
       {
         "offsets" : [ {
           "partition" : {
             "filename" : "/data/myfile.txt"
           },
           "offset" : {
             "position" : 15000 

   2

   
           }
         } ]
       }

   1

   Edits must be made within the offsets.json property.

   2

   The updated offset position in the source partition.

3. Run the command to update the offset position by annotating the KafkaConnector resource:

   oc annotate kafkaconnector my-source-connector strimzi.io/connector-offsets=alter -n <namespace>

   The annotation remains until either the update operation succeeds or it is manually removed from the resource.

4. Check the changes by using the procedure to list connector offsets.

5. Restart the connector by changing the state to running.

   Example configuration to start a connector

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     state: running
     # ...

Procedure

1. Edit the KafkaConnector resource to stop the connector.

   Example configuration to stop a connector

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     # ...
     state: stopped 

   1

   
     # ...

   1

   Changes the state of the connector to stopped. The default state for the connector when this property is not set is running.

2. Run the command to reset the offset position by annotating the KafkaConnector resource:

   oc annotate kafkaconnector my-source-connector strimzi.io/connector-offsets=reset -n <namespace>

   The annotation remains until either the reset operation succeeds or it is manually removed from the resource.

3. Check the changes by using the procedure to list connector offsets.

   After resetting, the offsets.json property is empty.

   Example source connector offset list

   apiVersion: v1
   kind: ConfigMap
   metadata:
     # ...
   data:
     offsets.json: |-
       {
         "offsets" : []
       }

4. Restart the connector by changing the state to running.

   Example configuration to start a connector

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     state: running
     # ...

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: 4.1.0
       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
     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: 4.1.0
       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
     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: 4.1.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
               pattern: "*.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
               pattern: "*.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. Supported authentication methods include tls, scram-sha-256, scram-sha-512, plain, and oauth.

   3

   The TLS certificates for the target Kafka cluster.

   4

   The user authentication for accessing the target Kafka cluster.

6. Apply the changes to the KafkaMirrorMaker2 resource to the same namespace as the target Kafka cluster.