Chapter 3. Deployment configuration

Procedure

1. In a Kafka resource, increase the size of the persistent volume allocated to the Kafka cluster, the ZooKeeper cluster, or both.

   • To increase the volume size allocated to the Kafka cluster, edit the spec.kafka.storage property.

   • To increase the volume size allocated to the ZooKeeper cluster, edit the spec.zookeeper.storage property.

     For example, to increase the volume size from 1000Gi to 2000Gi:

     apiVersion: kafka.strimzi.io/v1beta1
     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.

   Use oc apply:

   oc apply -f your-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.

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/v1beta1
   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.

   This can be done using oc apply:

   oc apply -f your-file

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

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.

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/v1beta1
   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.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the replicas property in the Kafka resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       replicas: 3
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Open the YAML configuration file that contains the Kafka resource specifying the cluster deployment.

2. In the spec.kafka.config property in the Kafka resource, enter one or more Kafka configuration settings. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       config:
         default.replication.factor: 3
         offsets.topic.replication.factor: 3
         transaction.state.log.replication.factor: 3
         transaction.state.log.min.isr: 1
       # ...
     zookeeper:
       # ...

3. Apply the new configuration to create or update the resource.

   Use oc apply:

   oc apply -f kafka.yaml

   where kafka.yaml is the YAML configuration file for the resource that you want to configure; for example, kafka-persistent.yaml.

Procedure

1. Edit the listeners property in the Kafka.spec.kafka resource.

   An example configuration of the plain (unencrypted) listener without authentication:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       listeners:
         plain: {}
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Open the YAML configuration file that contains the Kafka resource specifying the cluster deployment.

2. In the spec.kafka.listeners property in the Kafka resource, add the authentication field to the listeners for which you want to enable authentication. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       listeners:
         tls:
           authentication:
             type: tls
       # ...
     zookeeper:
       # ...

3. Apply the new configuration to create or update the resource.

   Use oc apply:

   oc apply -f kafka.yaml

   where kafka.yaml is the YAML configuration file for the resource that you want to configure; for example, kafka-persistent.yaml.

Procedure

1. Add or edit the authorization property in the Kafka.spec.kafka resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       authorization:
         type: simple
         superUsers:
           - CN=fred
           - sam
           - CN=edward
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Open the YAML configuration file that contains the Kafka resource specifying the cluster deployment.

2. In the spec.zookeeper.replicas property in the Kafka resource, enter the number of replicated ZooKeeper servers. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       replicas: 3
       # ...

3. Apply the new configuration to create or update the resource.

   Use oc apply:

   oc apply -f kafka.yaml

   where kafka.yaml is the YAML configuration file for the resource that you want to configure; for example, kafka-persistent.yaml.

Procedure

1. Open the YAML configuration file that contains the Kafka resource specifying the cluster deployment.

2. In the spec.zookeeper.config property in the Kafka resource, enter one or more ZooKeeper configuration settings. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       config:
         autopurge.snapRetainCount: 3
         autopurge.purgeInterval: 1
       # ...

3. Apply the new configuration to create or update the resource.

   Use oc apply:

   oc apply -f kafka.yaml

   where kafka.yaml is the YAML configuration file for the resource that you want to configure; for example, kafka-persistent.yaml.

Procedure

1. Open the terminal using the OpenShift console or run the exec command from your CLI.

   For example:

   oc exec -it my-cluster-kafka-0 -- bin/kafka-topics.sh --list --zookeeper localhost:2181

   Be sure to use localhost:2181.

   You can now run Kafka commands to ZooKeeper.

Procedure

1. Edit the entityOperator property in the Kafka resource. For example:

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

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

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

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       resources:
         requests:
           cpu: "8"
           memory: 64Gi
         limits:
           cpu: "12"
           memory: 128Gi
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Consult your OpenShift administrator regarding the node label that represents the zone / rack into which the node is deployed.

2. Edit the rack property in the Kafka resource using the label as the topology key.

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       rack:
         topologyKey: failure-domain.beta.kubernetes.io/zone
       # ...

3. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the livenessProbe or readinessProbe property in the Kafka, KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       readinessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the metrics property in the Kafka, KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       metrics:
         lowercaseOutputName: true
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the jvmOptions property in the Kafka, KafkaConnect, KafkaConnectS2I, KafkaMirrorMaker, or KafkaBridge resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       jvmOptions:
         "-Xmx": "8g"
         "-Xms": "8g"
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the image property in the Kafka, KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       image: my-org/my-image:latest
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the tlsSidecar property in the Kafka resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       tlsSidecar:
         resources:
           requests:
             cpu: 200m
             memory: 64Mi
           limits:
             cpu: 500m
             memory: 128Mi
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the KafkaExporter properties for the Kafka resource.

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

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     # ...
     kafkaExporter:
       image: my-org/my-image:latest 1
       groupRegex: ".*" 2
       topicRegex: ".*" 3
       resources: 4
         requests:
           cpu: 200m
           memory: 64Mi
         limits:
           cpu: 500m
           memory: 128Mi
       logging: debug 5
       enableSaramaLogging: true 6
       template: 7
         pod:
           metadata:
             labels:
               label1: value1
           imagePullSecrets:
             - name: my-docker-credentials
           securityContext:
             runAsUser: 1000001
             fsGroup: 0
           terminationGracePeriodSeconds: 120
       readinessProbe: 8
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe: 9
         initialDelaySeconds: 15
         timeoutSeconds: 5
   # ...

   1

   ADVANCED OPTION: Container image configuration, which is recommended only in special situations.

   2

   A regular expression to specify the consumer groups to include in the metrics.

   3

   A regular expression to specify the topics to include in the metrics.

   4

   CPU and memory resources to reserve.

   5

   Logging configuration, to log messages with a given severity (debug, info, warn, error, fatal) or above.

   6

   Boolean to enable Sarama logging, a Go client library used by Kafka Exporter.

   7

   Customization of deployment templates and pods.

   8

   Healthcheck readiness probes.

   9

   Healthcheck liveness probes.

2. Create or update the resource:

   oc apply -f kafka.yaml

Procedure

1. Prepare a JSON file named topics.json that lists the topics to move. It must have the following structure:

   {
     "version": 1,
     "topics": [
       <TopicObjects>
     ]
   }

   where <TopicObjects> is a comma-separated list of objects like:

   {
     "topic": <TopicName>
   }

   For example if you want to reassign all the partitions of topic-a and topic-b, you would need to prepare a topics.json file like this:

   {
     "version": 1,
     "topics": [
       { "topic": "topic-a"},
       { "topic": "topic-b"}
     ]
   }

2. Copy the topics.json file to one of the broker pods:

   cat topics.json | oc exec -c kafka <BrokerPod> -i -- \
     /bin/bash -c \
     'cat > /tmp/topics.json'

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

   oc exec <BrokerPod> -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --topics-to-move-json-file /tmp/topics.json \
     --broker-list <BrokerList> \
     --generate

   For example, to move all the partitions of topic-a and topic-b to brokers 4 and 7

   oc exec <BrokerPod> -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --topics-to-move-json-file /tmp/topics.json \
     --broker-list 4,7 \
     --generate

Procedure

1. Add as many new brokers as you need by increasing the Kafka.spec.kafka.replicas configuration option.
2. Verify that the new broker pods have started.

3. Copy the reassignment.json file to the broker pod on which you will later execute the commands:

   cat reassignment.json | \
     oc exec broker-pod -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

   For example:

   cat reassignment.json | \
     oc exec my-cluster-kafka-0 -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

4. Execute the partition reassignment using the kafka-reassign-partitions.sh command line tool from the same broker pod.

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --execute

   If you are going to throttle replication you can also pass the --throttle option with an inter-broker throttled rate in bytes per second. For example:

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 5000000 \
     --execute

   This command will print out two reassignment JSON objects. The first records the current assignment for the partitions being moved. You should save this to a local file (not a file in the pod) in case you need to revert the reassignment later on. The second JSON object is the target reassignment you have passed in your reassignment JSON file.

5. If you need to change the throttle during reassignment you can use the same command line with a different throttled rate. For example:

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 10000000 \
     --execute

6. Periodically verify whether the reassignment has completed using the kafka-reassign-partitions.sh command line tool from any of the broker pods. This is the same command as the previous step but with the --verify option instead of the --execute option.

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

   For example,

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

7. The reassignment has finished when the --verify command reports each of the partitions being moved as completed successfully. This final --verify will also have the effect of removing any reassignment throttles. You can now delete the revert file if you saved the JSON for reverting the assignment to their original brokers.

Procedure

1. Copy the reassignment.json file to the broker pod on which you will later execute the commands:

   cat reassignment.json | \
     oc exec broker-pod -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

   For example:

   cat reassignment.json | \
     oc exec my-cluster-kafka-0 -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

2. Execute the partition reassignment using the kafka-reassign-partitions.sh command line tool from the same broker pod.

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --execute

   If you are going to throttle replication you can also pass the --throttle option with an inter-broker throttled rate in bytes per second. For example:

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 5000000 \
     --execute

   This command will print out two reassignment JSON objects. The first records the current assignment for the partitions being moved. You should save this to a local file (not a file in the pod) in case you need to revert the reassignment later on. The second JSON object is the target reassignment you have passed in your reassignment JSON file.

3. If you need to change the throttle during reassignment you can use the same command line with a different throttled rate. For example:

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 10000000 \
     --execute

4. Periodically verify whether the reassignment has completed using the kafka-reassign-partitions.sh command line tool from any of the broker pods. This is the same command as the previous step but with the --verify option instead of the --execute option.

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

   For example,

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

5. The reassignment has finished when the --verify command reports each of the partitions being moved as completed successfully. This final --verify will also have the effect of removing any reassignment throttles. You can now delete the revert file if you saved the JSON for reverting the assignment to their original brokers.

6. Once all the partition reassignments have finished, the broker(s) being removed should not have responsibility for any of the partitions in the cluster. You can verify this by checking that the broker’s data log directory does not contain any live partition logs. If the log directory on the broker contains a directory that does not match the extended regular expression \.[a-z0-9]-delete$ then the broker still has live partitions and it should not be stopped.

   You can check this by executing the command:

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     /bin/bash -c \
     "ls -l /var/lib/kafka/kafka-log_<N>_ | grep -E '^d' | grep -vE '[a-zA-Z0-9.-]+\.[a-z0-9]+-delete$'"

   where N is the number of the Pod(s) being deleted.

   If the above command prints any output then the broker still has live partitions. In this case, either the reassignment has not finished, or the reassignment JSON file was incorrect.

7. Once you have confirmed that the broker has no live partitions you can edit the Kafka.spec.kafka.replicas of your Kafka resource, which will scale down the StatefulSet, deleting the highest numbered broker Pod(s).

Procedure

1. Add or edit the maintenanceTimeWindows property in the Kafka resource. For example to allow maintenance between 0800 and 1059 and between 1400 and 1559 you would set the maintenanceTimeWindows as shown below:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
     maintenanceTimeWindows:
       - "* * 8-10 * * ?"
       - "* * 14-15 * * ?"

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the replicas property in the KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnectS2I
   metadata:
     name: my-cluster
   spec:
     # ...
     replicas: 3
     # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the bootstrapServers property in the KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-cluster
   spec:
     # ...
     bootstrapServers: my-cluster-kafka-bootstrap:9092
     # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. (Optional) If they do not already exist, prepare the TLS certificate used in authentication in a file and create a Secret.

   Note

   The secrets created by the Cluster Operator for Kafka cluster may be used directly.

   This can be done using oc create:

   oc create secret generic my-secret --from-file=my-file.crt

2. Edit the tls property in the KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     tls:
       trustedCertificates:
         - secretName: my-cluster-cluster-cert
           certificate: ca.crt
     # ...

3. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. (Optional) If they do not already exist, prepare the keys used for authentication in a file and create the Secret.

   Note

   Secrets created by the User Operator may be used.

   This can be done using oc create:

   oc create secret generic my-secret --from-file=my-public.crt --from-file=my-private.key

2. Edit the authentication property in the KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     authentication:
       type: tls
       certificateAndKey:
         secretName: my-secret
         certificate: my-public.crt
         key: my-private.key
     # ...

3. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. (Optional) If they do not already exist, prepare a file with the password used in authentication and create the Secret.

   Note

   Secrets created by the User Operator may be used.

   This can be done using oc create:

   echo -n '<password>' > <my-password.txt>
   oc create secret generic <my-secret> --from-file=<my-password.txt>

2. Edit the authentication property in the KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     authentication:
       type: scram-sha-512
       username: <my-username>
       passwordSecret:
         secretName: <my-secret>
         password: <my-password.txt>
     # ...

3. Create or update the resource.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the config property in the KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     config:
       group.id: my-connect-cluster
       offset.storage.topic: my-connect-cluster-offsets
       config.storage.topic: my-connect-cluster-configs
       status.storage.topic: my-connect-cluster-status
       key.converter: org.apache.kafka.connect.json.JsonConverter
       value.converter: org.apache.kafka.connect.json.JsonConverter
       key.converter.schemas.enable: true
       value.converter.schemas.enable: true
       config.storage.replication.factor: 3
       offset.storage.replication.factor: 3
       status.storage.replication.factor: 3
     # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

3. If authorization is enabled for Kafka Connect, configure the Kafka Connect user to enable access to the Kafka Connect consumer group and topics.

Procedure

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

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

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

   apiVersion: kafka.strimzi.io/v1beta1
   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
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Read
           host: "*"
         # access to status.storage.topic
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Read
           host: "*"
         # access to config.storage.topic
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Read
           host: "*"
         # consumer group
         - resource:
             type: group
             name: connect-cluster
             patternType: literal
           operation: Read
           host: "*"

2. Create or update the resource.

   oc apply -f your-file

Procedure

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

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       resources:
         requests:
           cpu: "8"
           memory: 64Gi
         limits:
           cpu: "12"
           memory: 128Gi
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the livenessProbe or readinessProbe property in the Kafka, KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       readinessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the metrics property in the Kafka, KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       metrics:
         lowercaseOutputName: true
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the jvmOptions property in the Kafka, KafkaConnect, KafkaConnectS2I, KafkaMirrorMaker, or KafkaBridge resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       jvmOptions:
         "-Xmx": "8g"
         "-Xms": "8g"
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the image property in the Kafka, KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       image: my-org/my-image:latest
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Create a secret containing the information that will be mounted as an environment variable. For example:

   apiVersion: v1
   kind: Secret
   metadata:
     name: aws-creds
   type: Opaque
   data:
     awsAccessKey: QUtJQVhYWFhYWFhYWFhYWFg=
     awsSecretAccessKey: Ylhsd1lYTnpkMjl5WkE=

2. Create or edit the Kafka Connect resource. Configure the externalConfiguration section of the KafkaConnect or KafkaConnectS2I custom resource to reference the secret. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     externalConfiguration:
       env:
         - name: AWS_ACCESS_KEY_ID
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsAccessKey
         - name: AWS_SECRET_ACCESS_KEY
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsSecretAccessKey

3. Apply the changes to your Kafka Connect deployment.

   Use oc apply:

   oc apply -f your-file

Procedure

1. Create a secret containing a properties file that defines the configuration options for your connector configuration. For example:

   apiVersion: v1
   kind: Secret
   metadata:
     name: mysecret
   type: Opaque
   stringData:
     connector.properties: |-
       dbUsername: my-user
       dbPassword: my-password

2. Create or edit the Kafka Connect resource. Configure the FileConfigProvider in the config section and the externalConfiguration section of the KafkaConnect or KafkaConnectS2I custom resource to reference the secret. For example:

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

3. Apply the changes to your Kafka Connect deployment.

   Use oc apply:

   oc apply -f your-file

4. Configure your connector

   • If you are using the Kafka Connect HTTP REST interface, use the values from the mounted properties file in your JSON payload with connector configuration. For example:

     {
        "name":"my-connector",
        "config":{
           "connector.class":"MyDbConnector",
           "tasks.max":"3",
           "database": "my-postgresql:5432",
           "username":"${file:/opt/kafka/external-configuration/connector-config/connector.properties:dbUsername}",
           "password":"${file:/opt/kafka/external-configuration/connector-config/connector.properties:dbPassword}",
           # ...
        }
     }

   • If you are using a KafkaConnector resource, use the values from the mounted properties file in the spec.config section of your custom resource. For example:

     apiVersion: kafka.strimzi.io/v1beta1
     kind: KafkaConnector
     metadata:
       name: my-connector
       # ...
     spec:
       class: "MyDbConnector"
       tasksMax: 3
       config:
         database: "my-postgresql:5432"
         username: "${file:/opt/kafka/external-configuration/connector-config/connector.properties:dbUsername}"
         password: "${file:/opt/kafka/external-configuration/connector-config/connector.properties:dbPassword}"

Procedure

1. Edit the replicas property in the KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnectS2I
   metadata:
     name: my-cluster
   spec:
     # ...
     replicas: 3
     # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the bootstrapServers property in the KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-cluster
   spec:
     # ...
     bootstrapServers: my-cluster-kafka-bootstrap:9092
     # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. (Optional) If they do not already exist, prepare the TLS certificate used in authentication in a file and create a Secret.

   Note

   The secrets created by the Cluster Operator for Kafka cluster may be used directly.

   This can be done using oc create:

   oc create secret generic my-secret --from-file=my-file.crt

2. Edit the tls property in the KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     tls:
       trustedCertificates:
         - secretName: my-cluster-cluster-cert
           certificate: ca.crt
     # ...

3. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. (Optional) If they do not already exist, prepare the keys used for authentication in a file and create the Secret.

   Note

   Secrets created by the User Operator may be used.

   This can be done using oc create:

   oc create secret generic my-secret --from-file=my-public.crt --from-file=my-private.key

2. Edit the authentication property in the KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     authentication:
       type: tls
       certificateAndKey:
         secretName: my-secret
         certificate: my-public.crt
         key: my-private.key
     # ...

3. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. (Optional) If they do not already exist, prepare a file with the password used in authentication and create the Secret.

   Note

   Secrets created by the User Operator may be used.

   This can be done using oc create:

   echo -n '<password>' > <my-password.txt>
   oc create secret generic <my-secret> --from-file=<my-password.txt>

2. Edit the authentication property in the KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     authentication:
       type: scram-sha-512
       username: <my-username>
       passwordSecret:
         secretName: <my-secret>
         password: <my-password.txt>
     # ...

3. Create or update the resource.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the config property in the KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     config:
       group.id: my-connect-cluster
       offset.storage.topic: my-connect-cluster-offsets
       config.storage.topic: my-connect-cluster-configs
       status.storage.topic: my-connect-cluster-status
       key.converter: org.apache.kafka.connect.json.JsonConverter
       value.converter: org.apache.kafka.connect.json.JsonConverter
       key.converter.schemas.enable: true
       value.converter.schemas.enable: true
       config.storage.replication.factor: 3
       offset.storage.replication.factor: 3
       status.storage.replication.factor: 3
     # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

3. If authorization is enabled for Kafka Connect, configure the Kafka Connect user to enable access to the Kafka Connect consumer group and topics.

Procedure

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

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

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

   apiVersion: kafka.strimzi.io/v1beta1
   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
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Read
           host: "*"
         # access to status.storage.topic
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Read
           host: "*"
         # access to config.storage.topic
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Read
           host: "*"
         # consumer group
         - resource:
             type: group
             name: connect-cluster
             patternType: literal
           operation: Read
           host: "*"

2. Create or update the resource.

   oc apply -f your-file

Procedure

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

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       resources:
         requests:
           cpu: "8"
           memory: 64Gi
         limits:
           cpu: "12"
           memory: 128Gi
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the livenessProbe or readinessProbe property in the Kafka, KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       readinessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the metrics property in the Kafka, KafkaConnect or KafkaConnectS2I resource. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       metrics:
         lowercaseOutputName: true
       # ...

2. Create or update the resource.

   This can be done using oc apply:

   oc apply -f your-file