Deploying and Managing AMQ Streams on OpenShift

Procedure

1. Edit the AMQ Streams installation files to use the namespace the Cluster Operator is going to be installed into.

   For example, in this procedure the Cluster Operator is installed into the namespace my-cluster-operator-namespace.

   On Linux, use:

   sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

   On MacOS, use:

   sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

2. Deploy the Cluster Operator:

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

3. Check the status of the deployment:

   oc get deployments -n my-cluster-operator-namespace

   Output shows the deployment name and readiness

   NAME                      READY  UP-TO-DATE  AVAILABLE
   strimzi-cluster-operator  1/1    1           1

   READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1.

Procedure

1. Edit the AMQ Streams installation files to use the namespace the Cluster Operator is going to be installed into.

   For example, in this procedure the Cluster Operator is installed into the namespace my-cluster-operator-namespace.

   On Linux, use:

   sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

   On MacOS, use:

   sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

2. Edit the install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml file to add a list of all the namespaces the Cluster Operator will watch to the STRIMZI_NAMESPACE environment variable.

   For example, in this procedure the Cluster Operator will watch the namespaces watched-namespace-1, watched-namespace-2, watched-namespace-3.

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         serviceAccountName: strimzi-cluster-operator
         containers:
         - name: strimzi-cluster-operator
           image: registry.redhat.io/amq-streams/strimzi-rhel8-operator:2.5.2
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: watched-namespace-1,watched-namespace-2,watched-namespace-3

3. For each namespace listed, install the RoleBindings.

   In this example, we replace watched-namespace in these commands with the namespaces listed in the previous step, repeating them for watched-namespace-1, watched-namespace-2, watched-namespace-3:

   oc create -f install/cluster-operator/020-RoleBinding-strimzi-cluster-operator.yaml -n <watched_namespace>
   oc create -f install/cluster-operator/023-RoleBinding-strimzi-cluster-operator.yaml -n <watched_namespace>
   oc create -f install/cluster-operator/031-RoleBinding-strimzi-cluster-operator-entity-operator-delegation.yaml -n <watched_namespace>

4. Deploy the Cluster Operator:

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

5. Check the status of the deployment:

   oc get deployments -n my-cluster-operator-namespace

   Output shows the deployment name and readiness

   NAME                      READY  UP-TO-DATE  AVAILABLE
   strimzi-cluster-operator  1/1    1           1

   READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1.

Procedure

1. Edit the AMQ Streams installation files to use the namespace the Cluster Operator is going to be installed into.

   For example, in this procedure the Cluster Operator is installed into the namespace my-cluster-operator-namespace.

   On Linux, use:

   sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

   On MacOS, use:

   sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

2. Edit the install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml file to set the value of the STRIMZI_NAMESPACE environment variable to *.

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         # ...
         serviceAccountName: strimzi-cluster-operator
         containers:
         - name: strimzi-cluster-operator
           image: registry.redhat.io/amq-streams/strimzi-rhel8-operator:2.5.2
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: "*"
           # ...

3. Create ClusterRoleBindings that grant cluster-wide access for all namespaces to the Cluster Operator.

   oc create clusterrolebinding strimzi-cluster-operator-namespaced --clusterrole=strimzi-cluster-operator-namespaced --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator
   oc create clusterrolebinding strimzi-cluster-operator-watched --clusterrole=strimzi-cluster-operator-watched --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator
   oc create clusterrolebinding strimzi-cluster-operator-entity-operator-delegation --clusterrole=strimzi-entity-operator --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator

4. Deploy the Cluster Operator to your OpenShift cluster.

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

5. Check the status of the deployment:

   oc get deployments -n my-cluster-operator-namespace

   Output shows the deployment name and readiness

   NAME                      READY  UP-TO-DATE  AVAILABLE
   strimzi-cluster-operator  1/1    1           1

   READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1.

Procedure

1. Create and deploy an ephemeral or persistent cluster.

   • To create and deploy an ephemeral cluster:

     oc apply -f examples/kafka/kafka-ephemeral.yaml

   • To create and deploy a persistent cluster:

     oc apply -f examples/kafka/kafka-persistent.yaml

2. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Output shows the pod names and readiness

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   my-cluster-kafka-0          1/1     Running   0
   my-cluster-kafka-1          1/1     Running   0
   my-cluster-kafka-2          1/1     Running   0
   my-cluster-zookeeper-0      1/1     Running   0
   my-cluster-zookeeper-1      1/1     Running   0
   my-cluster-zookeeper-2      1/1     Running   0

   my-cluster is the name of the Kafka cluster.

   A sequential index number starting with 0 identifies each Kafka and ZooKeeper pod created.

   With the default deployment, you create an Entity Operator cluster, 3 Kafka pods, and 3 ZooKeeper pods.

   READY shows the number of replicas that are ready/expected. The deployment is successful when the STATUS displays as Running.

Procedure

1. Enable the KafkaNodePools feature gate from the command line:

   oc set env deployment/strimzi-cluster-operator STRIMZI_FEATURE_GATES="+KafkaNodePools"

   Or by editing the Cluster Operator Deployment and updating the STRIMZI_FEATURE_GATES environment variable:

   env
     - name: STRIMZI_FEATURE_GATES
       value: +KafkaNodePools

   This updates the Cluster Operator.

   If using KRaft mode, enable the UseKRaft feature gate as well.

2. Create a node pool.

   • To deploy a Kafka cluster and ZooKeeper cluster with two node pools of three brokers:

     oc apply -f examples/kafka/nodepools/kafka.yaml

   • To deploy a Kafka cluster in KRaft mode with a single node pool that uses dual-role nodes:

     oc apply -f examples/kafka/nodepools/kafka-with-dual-role-kraft-nodes.yaml

   • To deploy a Kafka cluster in KRaft mode with separate node pools for broker and controller nodes:

     oc apply -f examples/kafka/nodepools/kafka-with-kraft.yaml

3. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Output shows the node pool names and readiness

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   my-cluster-pool-a-kafka-0   1/1     Running   0
   my-cluster-pool-a-kafka-1   1/1     Running   0
   my-cluster-pool-a-kafka-4   1/1     Running   0

   • my-cluster is the name of the Kafka cluster.

   • pool-a is the name of the node pool.

     A sequential index number starting with 0 identifies each Kafka pod created. If you are using ZooKeeper, you’ll also see the ZooKeeper pods.

     READY shows the number of replicas that are ready/expected. The deployment is successful when the STATUS displays as Running.

     Information on the deployment is also shown in the status of the KafkaNodePool resource, including a list of IDs for nodes in the pool.

     Note

     Node IDs are assigned sequentially starting at 0 (zero) across all node pools within a cluster. This means that node IDs might not run sequentially within a specific node pool. If there are gaps in the sequence of node IDs across the cluster, the next node to be added is assigned an ID that fills the gap. When scaling down, the node with the highest node ID within a pool is removed.

Procedure

1. Edit the entityOperator properties of the Kafka resource to include topicOperator:

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

2. Configure the Topic Operator spec using the properties described in the EntityTopicOperatorSpec schema reference.

   Use an empty object ({}) if you want all properties to use their default values.

3. Create or update the resource:

   oc apply -f <kafka_configuration_file>

4. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Output shows the pod name and readiness

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   # ...

   my-cluster is the name of the Kafka cluster.

   READY shows the number of replicas that are ready/expected. The deployment is successful when the STATUS displays as Running.

Procedure

1. Edit the entityOperator properties of the Kafka resource to include userOperator:

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

2. Configure the User Operator spec using the properties described in EntityUserOperatorSpec schema reference.

   Use an empty object ({}) if you want all properties to use their default values.

3. Create or update the resource:

   oc apply -f <kafka_configuration_file>

4. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Output shows the pod name and readiness

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   # ...

   my-cluster is the name of the Kafka cluster.

   READY shows the number of replicas that are ready/expected. The deployment is successful when the STATUS displays as Running.

Procedure

1. Deploy Kafka Connect to your OpenShift cluster. Use the examples/connect/kafka-connect.yaml file to deploy Kafka Connect.

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

2. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Output shows the deployment name and readiness

   NAME                                 READY  STATUS   RESTARTS
   my-connect-cluster-connect-<pod_id>  1/1    Running  0

   my-connect-cluster is the name of the Kafka Connect cluster.

   A pod ID identifies each pod created.

   With the default deployment, you create a single Kafka Connect pod.

   READY shows the number of replicas that are ready/expected. The deployment is successful when the STATUS displays as Running.

Procedure

1. Deploy Kafka MirrorMaker to your OpenShift cluster:

   For MirrorMaker:

   oc apply -f examples/mirror-maker/kafka-mirror-maker.yaml

   For MirrorMaker 2:

   oc apply -f examples/mirror-maker/kafka-mirror-maker-2.yaml

2. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Output shows the deployment name and readiness

   NAME                                    READY  STATUS   RESTARTS
   my-mirror-maker-mirror-maker-<pod_id>   1/1    Running  1
   my-mm2-cluster-mirrormaker2-<pod_id>    1/1    Running  1

   my-mirror-maker is the name of the Kafka MirrorMaker cluster. my-mm2-cluster is the name of the Kafka MirrorMaker 2 cluster.

   A pod ID identifies each pod created.

   With the default deployment, you install a single MirrorMaker or MirrorMaker 2 pod.

   READY shows the number of replicas that are ready/expected. The deployment is successful when the STATUS displays as Running.

Procedure

1. Deploy Kafka Bridge to your OpenShift cluster:

   oc apply -f examples/bridge/kafka-bridge.yaml

2. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Output shows the deployment name and readiness

   NAME                       READY  STATUS   RESTARTS
   my-bridge-bridge-<pod_id>  1/1    Running  0

   my-bridge is the name of the Kafka Bridge cluster.

   A pod ID identifies each pod created.

   With the default deployment, you install a single Kafka Bridge pod.

   READY shows the number of replicas that are ready/expected. The deployment is successful when the STATUS displays as Running.

Procedure

1. List the names of the pods in your OpenShift cluster:

   oc get pods -o name
   
   pod/kafka-consumer
   # ...
   pod/my-bridge-bridge-<pod_id>

2. Connect to the Kafka Bridge pod on port 8080:

   oc port-forward pod/my-bridge-bridge-<pod_id> 8080:8080 &

   Note

   If port 8080 on your local machine is already in use, use an alternative HTTP port, such as 8008.

Procedure

1. Edit the env properties in the install/topic-operator/05-Deployment-strimzi-topic-operator.yaml standalone deployment file.

   Example standalone Topic Operator deployment configuration

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: strimzi-topic-operator
     labels:
       app: strimzi
   spec:
     # ...
     template:
       # ...
       spec:
         # ...
         containers:
           - name: strimzi-topic-operator
             # ...
             env:
               - name: STRIMZI_NAMESPACE 1
                 valueFrom:
                   fieldRef:
                     fieldPath: metadata.namespace
               - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 2
                 value: my-kafka-bootstrap-address:9092
               - name: STRIMZI_RESOURCE_LABELS 3
                 value: "strimzi.io/cluster=my-cluster"
               - name: STRIMZI_ZOOKEEPER_CONNECT 4
                 value: my-cluster-zookeeper-client:2181
               - name: STRIMZI_ZOOKEEPER_SESSION_TIMEOUT_MS 5
                 value: "18000"
               - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 6
                 value: "120000"
               - name: STRIMZI_TOPIC_METADATA_MAX_ATTEMPTS 7
                 value: "6"
               - name: STRIMZI_LOG_LEVEL 8
                 value: INFO
               - name: STRIMZI_TLS_ENABLED 9
                 value: "false"
               - name: STRIMZI_JAVA_OPTS 10
                 value: "-Xmx=512M -Xms=256M"
               - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 11
                 value: "-Djavax.net.debug=verbose -DpropertyName=value"
               - name: STRIMZI_PUBLIC_CA 12
                 value: "false"
               - name: STRIMZI_TLS_AUTH_ENABLED 13
                 value: "false"
               - name: STRIMZI_SASL_ENABLED 14
                 value: "false"
               - name: STRIMZI_SASL_USERNAME 15
                 value: "admin"
               - name: STRIMZI_SASL_PASSWORD 16
                 value: "password"
               - name: STRIMZI_SASL_MECHANISM 17
                 value: "scram-sha-512"
               - name: STRIMZI_SECURITY_PROTOCOL 18
                 value: "SSL"

   1

   The OpenShift namespace for the Topic Operator to watch for KafkaTopic resources. Specify the namespace of the Kafka cluster.

   2

   The host and port pair of the bootstrap broker address to discover and connect to all brokers in the Kafka cluster. Use a comma-separated list to specify two or three broker addresses in case a server is down.

   3

   The label to identify the KafkaTopic resources managed by the Topic Operator. This does not have to be the name of the Kafka cluster. It can be the label assigned to the KafkaTopic resource. If you deploy more than one Topic Operator, the labels must be unique for each. That is, the operators cannot manage the same resources.

   4

   (ZooKeeper) The host and port pair of the address to connect to the ZooKeeper cluster. This must be the same ZooKeeper cluster that your Kafka cluster is using.

   5

   (ZooKeeper) The ZooKeeper session timeout, in milliseconds. The default is 18000 (18 seconds).

   6

   The interval between periodic reconciliations, in milliseconds. The default is 120000 (2 minutes).

   7

   The number of attempts at getting topic metadata from Kafka. The time between each attempt is defined as an exponential backoff. Consider increasing this value when topic creation takes more time due to the number of partitions or replicas. The default is 6 attempts.

   8

   The level for printing logging messages. You can set the level to ERROR, WARNING, INFO, DEBUG, or TRACE.

   9

   Enables TLS support for encrypted communication with the Kafka brokers.

   10

   (Optional) The Java options used by the JVM running the Topic Operator.

   11

   (Optional) The debugging (-D) options set for the Topic Operator.

   12

   (Optional) Skips the generation of trust store certificates if TLS is enabled through STRIMZI_TLS_ENABLED. If this environment variable is enabled, the brokers must use a public trusted certificate authority for their TLS certificates. The default is false.

   13

   (Optional) Generates key store certificates for mTLS authentication. Setting this to false disables client authentication with mTLS to the Kafka brokers. The default is true.

   14

   (Optional) Enables SASL support for client authentication when connecting to Kafka brokers. The default is false.

   15

   (Optional) The SASL username for client authentication. Mandatory only if SASL is enabled through STRIMZI_SASL_ENABLED.

   16

   (Optional) The SASL password for client authentication. Mandatory only if SASL is enabled through STRIMZI_SASL_ENABLED.

   17

   (Optional) The SASL mechanism for client authentication. Mandatory only if SASL is enabled through STRIMZI_SASL_ENABLED. You can set the value to plain, scram-sha-256, or scram-sha-512.

   18

   (Optional) The security protocol used for communication with Kafka brokers. The default value is "PLAINTEXT". You can set the value to PLAINTEXT, SSL, SASL_PLAINTEXT, or SASL_SSL.

2. If you want to connect to Kafka brokers that are using certificates from a public certificate authority, set STRIMZI_PUBLIC_CA to true. Set this property to true, for example, if you are using Amazon AWS MSK service.

3. If you enabled mTLS with the STRIMZI_TLS_ENABLED environment variable, specify the keystore and truststore used to authenticate connection to the Kafka cluster.

   Example mTLS configuration

   # ....
   env:
     - name: STRIMZI_TRUSTSTORE_LOCATION 1
       value: "/path/to/truststore.p12"
     - name: STRIMZI_TRUSTSTORE_PASSWORD 2
       value: "TRUSTSTORE-PASSWORD"
     - name: STRIMZI_KEYSTORE_LOCATION 3
       value: "/path/to/keystore.p12"
     - name: STRIMZI_KEYSTORE_PASSWORD 4
       value: "KEYSTORE-PASSWORD"
   # ...

   1

   The truststore contains the public keys of the Certificate Authorities used to sign the Kafka and ZooKeeper server certificates.

   2

   The password for accessing the truststore.

   3

   The keystore contains the private key for mTLS authentication.

   4

   The password for accessing the keystore.

4. Deploy the Topic Operator.

   oc create -f install/topic-operator

5. Check the status of the deployment:

   oc get deployments

   Output shows the deployment name and readiness

   NAME                    READY  UP-TO-DATE  AVAILABLE
   strimzi-topic-operator  1/1    1           1

   READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1.

Procedure

1. Edit the following env properties in the install/user-operator/05-Deployment-strimzi-user-operator.yaml standalone deployment file.

   Example standalone User Operator deployment configuration

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: strimzi-user-operator
     labels:
       app: strimzi
   spec:
     # ...
     template:
       # ...
       spec:
         # ...
         containers:
           - name: strimzi-user-operator
             # ...
             env:
               - name: STRIMZI_NAMESPACE 1
                 valueFrom:
                   fieldRef:
                     fieldPath: metadata.namespace
               - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 2
                 value: my-kafka-bootstrap-address:9092
               - name: STRIMZI_CA_CERT_NAME 3
                 value: my-cluster-clients-ca-cert
               - name: STRIMZI_CA_KEY_NAME 4
                 value: my-cluster-clients-ca
               - name: STRIMZI_LABELS 5
                 value: "strimzi.io/cluster=my-cluster"
               - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 6
                 value: "120000"
               - name: STRIMZI_WORK_QUEUE_SIZE 7
                 value: 10000
               - name: STRIMZI_CONTROLLER_THREAD_POOL_SIZE 8
                 value: 10
               - name: STRIMZI_USER_OPERATIONS_THREAD_POOL_SIZE 9
                 value: 4
               - name: STRIMZI_LOG_LEVEL 10
                 value: INFO
               - name: STRIMZI_GC_LOG_ENABLED 11
                 value: "true"
               - name: STRIMZI_CA_VALIDITY 12
                 value: "365"
               - name: STRIMZI_CA_RENEWAL 13
                 value: "30"
               - name: STRIMZI_JAVA_OPTS 14
                 value: "-Xmx=512M -Xms=256M"
               - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 15
                 value: "-Djavax.net.debug=verbose -DpropertyName=value"
               - name: STRIMZI_SECRET_PREFIX 16
                 value: "kafka-"
               - name: STRIMZI_ACLS_ADMIN_API_SUPPORTED 17
                 value: "true"
               - name: STRIMZI_MAINTENANCE_TIME_WINDOWS 18
                 value: '* * 8-10 * * ?;* * 14-15 * * ?'
               - name: STRIMZI_KAFKA_ADMIN_CLIENT_CONFIGURATION 19
                 value: |
                   default.api.timeout.ms=120000
                   request.timeout.ms=60000

   1

   The OpenShift namespace for the User Operator to watch for KafkaUser resources. Only one namespace can be specified.

   2

   The host and port pair of the bootstrap broker address to discover and connect to all brokers in the Kafka cluster. Use a comma-separated list to specify two or three broker addresses in case a server is down.

   3

   The OpenShift Secret that contains the public key (ca.crt) value of the CA (certificate authority) that signs new user certificates for mTLS authentication.

   4

   The OpenShift Secret that contains the private key (ca.key) value of the CA that signs new user certificates for mTLS authentication.

   5

   The label to identify the KafkaUser resources managed by the User Operator. This does not have to be the name of the Kafka cluster. It can be the label assigned to the KafkaUser resource. If you deploy more than one User Operator, the labels must be unique for each. That is, the operators cannot manage the same resources.

   6

   The interval between periodic reconciliations, in milliseconds. The default is 120000 (2 minutes).

   7

   The size of the controller event queue. The size of the queue should be at least as big as the maximal amount of users you expect the User Operator to operate. The default is 1024.

   8

   The size of the worker pool for reconciling the users. Bigger pool might require more resources, but it will also handle more KafkaUser resources The default is 50.

   9

   The size of the worker pool for Kafka Admin API and OpenShift operations. Bigger pool might require more resources, but it will also handle more KafkaUser resources The default is 4.

   10

   The level for printing logging messages. You can set the level to ERROR, WARNING, INFO, DEBUG, or TRACE.

   11

   Enables garbage collection (GC) logging. The default is true.

   12

   The validity period for the CA. The default is 365 days.

   13

   The renewal period for the CA. The renewal period is measured backwards from the expiry date of the current certificate. The default is 30 days to initiate certificate renewal before the old certificates expire.

   14

   (Optional) The Java options used by the JVM running the User Operator

   15

   (Optional) The debugging (-D) options set for the User Operator

   16

   (Optional) Prefix for the names of OpenShift secrets created by the User Operator.

   17

   (Optional) Indicates whether the Kafka cluster supports management of authorization ACL rules using the Kafka Admin API. When set to false, the User Operator will reject all resources with simple authorization ACL rules. This helps to avoid unnecessary exceptions in the Kafka cluster logs. The default is true.

   18

   (Optional) Semi-colon separated list of Cron Expressions defining the maintenance time windows during which the expiring user certificates will be renewed.

   19

   (Optional) Configuration options for configuring the Kafka Admin client used by the User Operator in the properties format.

2. If you are using mTLS to connect to the Kafka cluster, specify the secrets used to authenticate connection. Otherwise, go to the next step.

   Example mTLS configuration

   # ....
   env:
     - name: STRIMZI_CLUSTER_CA_CERT_SECRET_NAME 1
       value: my-cluster-cluster-ca-cert
     - name: STRIMZI_EO_KEY_SECRET_NAME 2
       value: my-cluster-entity-operator-certs
   # ..."

   1

   The OpenShift Secret that contains the public key (ca.crt) value of the CA that signs Kafka broker certificates.

   2

   The OpenShift Secret that contains the certificate public key (entity-operator.crt) and private key (entity-operator.key) that is used for mTLS authentication against the Kafka cluster.

3. Deploy the User Operator.

   oc create -f install/user-operator

4. Check the status of the deployment:

   oc get deployments

   Output shows the deployment name and readiness

   NAME                   READY  UP-TO-DATE  AVAILABLE
   strimzi-user-operator  1/1    1           1

   READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1.

Procedure

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

   The plugin properties are shown in this example configuration.

   Example Kafka Static Quota plugin configuration

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

   1

   Loads the Kafka Static Quota plugin.

   2

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

   3

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

   4

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

   5

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

   6

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

2. Update the resource.

   oc apply -f <kafka_configuration_file>

Procedure

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

2. You can now scale the node pool.

   For more information, see the following:

   • Section 8.3.2, “(Preview) Adding nodes to a node pool”
   • Section 8.3.3, “(Preview) Removing nodes from a node pool”
   • Section 8.3.4, “(Preview) Moving nodes between node pools”

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

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 have a status of READY.

   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-kafka-0  1/1    Running  0
   my-cluster-pool-a-kafka-1  1/1    Running  0
   my-cluster-pool-a-kafka-2  1/1    Running  0
   my-cluster-pool-a-kafka-3  1/1    Running  0

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

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

Procedure

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

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

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 have a status of READY.

   oc get pods -n <my_cluster_operator_namespace>

   Output shows four Kafka nodes in the target node pool

   NAME                       READY  STATUS   RESTARTS
   my-cluster-pool-a-kafka-0  1/1    Running  0
   my-cluster-pool-a-kafka-1  1/1    Running  0
   my-cluster-pool-a-kafka-4  1/1    Running  0
   my-cluster-pool-a-kafka-5  1/1    Running  0

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

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

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

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 within a 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-6  1/1    Running  0

Procedure

1. Create a new KafkaNodePool resource.

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

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

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

2. Apply the KafkaNodePool resource:

   oc apply -f <node_pool_configuration_file>

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

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

3. Update the STRIMZI_FEATURE_GATES environment variable in the Cluster Operator configuration to include +KafkaNodePools.

   env:
     - name: STRIMZI_FEATURE_GATES
       value: +KafkaNodePools

4. Enable the KafkaNodePools feature gate in the Kafka resource using the strimzi.io/node-pools: enabled annotation.

   Example configuration for a node pool in a cluster using ZooKeeper

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
     annotations:
       strimzi.io/node-pools: enabled
   spec:
     kafka:
       version: 3.5.0
       replicas: 3
     # ...
     storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false

5. Apply the Kafka resource:

   oc apply -f <kafka_configuration_file>

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.

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

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

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

2. Create or update the resource.

   oc apply -f KAFKA-USER-CONFIG-FILE

Procedure

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

   You can add listener configuration for authentication and enable authorization.

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

   Example source Kafka cluster configuration with TLS encryption and mTLS authentication

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

   Example target Kafka cluster configuration with TLS encryption and mTLS authentication

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

2. Create or update the Kafka resources in separate namespaces.

   oc apply -f <kafka_configuration_file> -n <namespace>

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

   The certificates are created in the secret <cluster_name>-cluster-ca-cert.

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

   1. Configure the same authentication and authorization types as the corresponding source and target Kafka cluster. For example, if you used tls authentication and the simple authorization type in the Kafka configuration for the source Kafka cluster, use the same in the KafkaUser configuration.

   2. Configure the ACLs needed by MirrorMaker 2 to allow operations on the source and target Kafka clusters.

      The ACLs are used by the internal MirrorMaker connectors, and by the underlying Kafka Connect framework.

   Example source user configuration for mTLS authentication

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

   Example target user configuration for mTLS authentication

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

   Note

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

4. Create or update a KafkaUser resource in each of the namespaces you created for the source and target Kafka clusters.

   oc apply -f <kafka_user_configuration_file> -n <namespace>

   The User Operator creates the users representing the client (MirrorMaker), and the security credentials used for client authentication, based on the chosen authentication type.

   The User Operator creates a new secret with the same name as the KafkaUser resource. The secret contains a private and public key for mTLS authentication. The public key is contained in a user certificate, which is signed by the clients CA.

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

   Example MirrorMaker 2 configuration with TLS encryption and mTLS authentication

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

   1

   The TLS certificates for the source Kafka cluster. If they are in a separate namespace, copy the cluster secrets from the namespace of the Kafka cluster.

   2

   The user authentication for accessing the source Kafka cluster using the TLS mechanism.

   3

   The TLS certificates for the target Kafka cluster.

   4

   The user authentication for accessing the target Kafka cluster.

6. Create or update the KafkaMirrorMaker2 resource in the same namespace as the target Kafka cluster.

   oc apply -f <mirrormaker2_configuration_file> -n <namespace_of_target_cluster>