Using AMQ Streams on OpenShift Container Platform

Procedure

1. Create a new Dockerfile using registry.redhat.io/amq7/amqstreams-kafka-22 as the base image:

   FROM registry.redhat.io/amq7/amqstreams-kafka-22
   USER root:root
   COPY ./my-plugins/ /opt/kafka/plugins/
   USER kafka:kafka

2. Build the container image.
3. Push your custom image to your container registry.

4. Edit the KafkaConnect.spec.image property of the KafkaConnect custom resource to point to the new container image. If set, this property overrides the STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE variable referred to in the next step.

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
   spec:
     #...
     image: my-new-container-image

5. In the install/cluster-operator/050-Deployment-strimzi-cluster-operator.yaml file, edit the STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE variable to point to the new container image.

Procedure

1. On the command line, use the oc apply command to create and deploy a Kafka Connect S2I cluster:

   oc apply -f examples/kafka-connect/kafka-connect-s2i.yaml

2. Create a directory with Kafka Connect plug-ins:

   $ tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   │   ├── bson-3.4.2.jar
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mongodb-0.7.1.jar
   │   ├── debezium-core-0.7.1.jar
   │   ├── LICENSE.txt
   │   ├── mongodb-driver-3.4.2.jar
   │   ├── mongodb-driver-core-3.4.2.jar
   │   └── README.md
   ├── debezium-connector-mysql
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mysql-0.7.1.jar
   │   ├── debezium-core-0.7.1.jar
   │   ├── LICENSE.txt
   │   ├── mysql-binlog-connector-java-0.13.0.jar
   │   ├── mysql-connector-java-5.1.40.jar
   │   ├── README.md
   │   └── wkb-1.0.2.jar
   └── debezium-connector-postgres
       ├── CHANGELOG.md
       ├── CONTRIBUTE.md
       ├── COPYRIGHT.txt
       ├── debezium-connector-postgres-0.7.1.jar
       ├── debezium-core-0.7.1.jar
       ├── LICENSE.txt
       ├── postgresql-42.0.0.jar
       ├── protobuf-java-2.6.1.jar
       └── README.md

3. Use the oc start-build command to start a new build of the image using the prepared directory:

   oc start-build my-connect-cluster-connect --from-dir ./my-plugins/

   Note

   The name of the build is the same as the name of the deployed Kafka Connect cluster.

4. Once the build has finished, the new image is used automatically by the Kafka Connect deployment.

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.

   On OpenShift, 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift, 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.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

Procedure

1. Deploy Kafka cluster with an external listener enabled and configured to the type route.

   An example configuration with an external listener configured to use Routes:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       listeners:
         external:
           type: route
           # ...
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   oc apply -f your-file

3. Find the address of the bootstrap Route.

   oc get routes _cluster-name_-kafka-bootstrap -o=jsonpath='{.status.ingress[0].host}{"\n"}'

   Use the address together with port 443 in your Kafka client as the bootstrap address.

4. Extract the public certificate of the broker certification authority

   oc extract secret/_cluster-name_-cluster-ca-cert --keys=ca.crt --to=- > ca.crt

   Use the extracted certificate in your Kafka client to configure TLS connection. If you enabled any authentication, you will also need to configure SASL or TLS authentication.

Procedure

1. Deploy Kafka cluster with an external listener enabled and configured to the type loadbalancer.

   An example configuration with an external listener configured to use loadbalancers:

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

2. Create or update the resource.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

3. Find the hostname of the bootstrap loadbalancer.

   On OpenShift this can be done using oc get:

   oc get service cluster-name-kafka-external-bootstrap -o=jsonpath='{.status.loadBalancer.ingress[0].hostname}{"\n"}'

   If no hostname was found (nothing was returned by the command), use the loadbalancer IP address.

   On OpenShift this can be done using oc get:

   oc get service cluster-name-kafka-external-bootstrap -o=jsonpath='{.status.loadBalancer.ingress[0].ip}{"\n"}'

   Use the hostname or IP address together with port 9094 in your Kafka client as the bootstrap address.

4. Unless TLS encryption was disabled, extract the public certificate of the broker certification authority.

   On OpenShift this can be done using oc extract:

   oc extract secret/cluster-name-cluster-ca-cert --keys=ca.crt --to=- > ca.crt

   Use the extracted certificate in your Kafka client to configure TLS connection. If you enabled any authentication, you will also need to configure SASL or TLS authentication.

Procedure

1. Deploy Kafka cluster with an external listener enabled and configured to the type nodeport.

   An example configuration with an external listener configured to use node ports:

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

2. Create or update the resource.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

3. Find the port number of the bootstrap service.

   On OpenShift this can be done using oc get:

   oc get service cluster-name-kafka-external-bootstrap -o=jsonpath='{.spec.ports[0].nodePort}{"\n"}'

   The port should be used in the Kafka bootstrap address.

4. Find the address of the OpenShift node.

   On OpenShift this can be done using oc get:

   oc get node node-name -o=jsonpath='{range .status.addresses[*]}{.type}{"\t"}{.address}{"\n"}'

   If several different addresses are returned, select the address type you want based on the following order:

   1. ExternalDNS
   2. ExternalIP
   3. Hostname
   4. InternalDNS

   5. InternalIP

      Use the address with the port found in the previous step in the Kafka bootstrap address.

5. Unless TLS encryption was disabled, extract the public certificate of the broker certification authority.

   On OpenShift this can be done using oc extract:

   oc extract secret/cluster-name-cluster-ca-cert --keys=ca.crt --to=- > ca.crt

   Use the extracted certificate in your Kafka client to configure TLS connection. If you enabled any authentication, you will also need to configure SASL or TLS authentication.

Procedure

1. Open the Kafka resource.

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

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

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       listeners:
         tls:
           networkPolicyPeers:
             - podSelector:
                 matchLabels:
                   app: kafka-client
       # ...
     zookeeper:
       # ...

3. Create or update the resource.

   On OpenShift use 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.

   On OpenShift, 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
       # ...
     zookeeper:
       # ...

2. Create or update the resource.

   On OpenShift 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.

   On OpenShift, 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.

   On OpenShift, 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 -ti 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.

   On OpenShift 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.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the YAML file to specify the loggers and logging level for the required components.

   For example, the logging level here is set to INFO:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       logging:
         type: inline
         loggers:
           logger.name: "INFO"
       # ...
     zookeeper:
       # ...
       logging:
         type: inline
         loggers:
           logger.name: "INFO"
       # ...
     entityOperator:
       # ...
       topicOperator:
         # ...
         logging:
           type: inline
           loggers:
             logger.name: "INFO"
       # ...
       # ...
       userOperator:
         # ...
         logging:
           type: inline
           loggers:
             logger.name: "INFO"
       # ...

   You can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.

   For more information about the log levels, see the log4j manual.

2. Create or update the Kafka resource in OpenShift.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the YAML file to specify the name of the ConfigMap to use for the required components. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       logging:
         type: external
         name: customConfigMap
       # ...

   Remember to place your custom ConfigMap under the log4j.properties or log4j2.properties key.

2. Create or update the Kafka resource in OpenShift.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the jvmOptions property in the Kafka, KafkaConnect or KafkaConnectS2I 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

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:

   On OpenShift:

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

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

   On OpenShift:

   oc rsh -c kafka <BrokerPod> \
     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 rsh -c kafka _<BrokerPod>_ \
     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:

   On OpenShift:

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

   For example:

   cat reassignment.json | \
     oc rsh -c kafka my-cluster-kafka-0 /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.

   On OpenShift:

   oc rsh -c kafka broker-pod \
     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:

   On OpenShift:

   oc rsh -c kafka my-cluster-kafka-0 \
     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:

   On OpenShift:

   oc rsh -c kafka my-cluster-kafka-0 \
     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.

   On OpenShift:

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

   For example, on OpenShift,

   oc rsh -c kafka my-cluster-kafka-0 \
     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:

   On OpenShift:

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

   For example:

   cat reassignment.json | \
     oc rsh -c kafka my-cluster-kafka-0 /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.

   On OpenShift:

   oc rsh -c kafka broker-pod \
     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:

   On OpenShift:

   oc rsh -c kafka my-cluster-kafka-0 \
     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:

   On OpenShift:

   oc rsh -c kafka my-cluster-kafka-0 \
     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.

   On OpenShift:

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

   For example, on OpenShift,

   oc rsh -c kafka my-cluster-kafka-0 \
     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 rsh <BrokerN> -c kafka /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.

   On OpenShift, use 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift this can be done using oc create:

   echo -n '1f2d1e2e67df' > <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.

   On OpenShift 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.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the YAML file to specify the loggers and logging level for the required components.

   For example, the logging level here is set to INFO:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   spec:
     # ...
     logging:
       type: inline
       loggers:
         logger.name: "INFO"
     # ...

   You can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.

   For more information about the log levels, see the log4j manual.

2. Create or update the Kafka resource in OpenShift.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the YAML file to specify the name of the ConfigMap to use for the required components. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   spec:
     # ...
     logging:
       type: external
       name: customConfigMap
     # ...

   Remember to place your custom ConfigMap under the log4j.properties or log4j2.properties key.

2. Create or update the Kafka resource in OpenShift.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the jvmOptions property in the Kafka, KafkaConnect or KafkaConnectS2I 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift use oc apply:

   oc apply -f your-file

4. 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}",
         # ...
      }
   }

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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift this can be done using oc create:

   echo -n '1f2d1e2e67df' > <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.

   On OpenShift 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.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the YAML file to specify the loggers and logging level for the required components.

   For example, the logging level here is set to INFO:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnectS2I
   spec:
     # ...
     logging:
       type: inline
       loggers:
         logger.name: "INFO"
     # ...

   You can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.

   For more information about the log levels, see the log4j manual.

2. Create or update the Kafka resource in OpenShift.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the YAML file to specify the name of the ConfigMap to use for the required components. For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnectS2I
   spec:
     # ...
     logging:
       type: external
       name: customConfigMap
     # ...

   Remember to place your custom ConfigMap under the log4j.properties or log4j2.properties key.

2. Create or update the Kafka resource in OpenShift.

   On OpenShift 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.

   On OpenShift 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.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file

Procedure

1. Edit the jvmOptions property in the Kafka, KafkaConnect or KafkaConnectS2I 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.

   On OpenShift 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.

   On OpenShift this can be done using oc apply:

   oc apply -f your-file