Chapter 8. Deploying Streams for Apache Kafka using installation files

Procedure

1. Edit the Streams for Apache Kafka 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

   Copy to Clipboard Toggle word wrap

   On MacOS, use:

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

   Copy to Clipboard Toggle word wrap

2. Deploy the Cluster Operator:

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

   Copy to Clipboard Toggle word wrap

3. Check the status of the deployment:

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

   Copy to Clipboard Toggle word wrap

   Output shows the deployment name and readiness

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

   Copy to Clipboard Toggle word wrap

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

Procedure

1. Edit the Streams for Apache Kafka 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

   Copy to Clipboard Toggle word wrap

   On MacOS, use:

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

   Copy to Clipboard Toggle word wrap

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-rhel9-operator:2.9.3
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: watched-namespace-1,watched-namespace-2,watched-namespace-3

   Copy to Clipboard Toggle word wrap

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>

   Copy to Clipboard Toggle word wrap

4. Deploy the Cluster Operator:

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

   Copy to Clipboard Toggle word wrap

5. Check the status of the deployment:

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

   Copy to Clipboard Toggle word wrap

   Output shows the deployment name and readiness

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

   Copy to Clipboard Toggle word wrap

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

Procedure

1. Edit the Streams for Apache Kafka 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

   Copy to Clipboard Toggle word wrap

   On MacOS, use:

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

   Copy to Clipboard Toggle word wrap

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-rhel9-operator:2.9.3
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: "*"
           # ...

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

4. Deploy the Cluster Operator to your OpenShift cluster.

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

   Copy to Clipboard Toggle word wrap

5. Check the status of the deployment:

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

   Copy to Clipboard Toggle word wrap

   Output shows the deployment name and readiness

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

   Copy to Clipboard Toggle word wrap

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

Procedure

1. Deploy a KRaft-based Kafka cluster.

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

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

   Copy to Clipboard Toggle word wrap

2. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

   Output shows the node pool names and readiness

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

   Copy to Clipboard Toggle word wrap

   • 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. Deploy a ZooKeeper-based Kafka cluster.

   • To deploy an ephemeral cluster:

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

     Copy to Clipboard Toggle word wrap

   • To deploy a persistent cluster:

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

     Copy to Clipboard Toggle word wrap

2. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

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>

   Copy to Clipboard Toggle word wrap

4. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

   Output shows the pod name and readiness

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

   Copy to Clipboard Toggle word wrap

   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: {}

   Copy to Clipboard Toggle word wrap

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>

   Copy to Clipboard Toggle word wrap

4. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

   Output shows the pod name and readiness

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

   Copy to Clipboard Toggle word wrap

   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. Open the terminal using the OpenShift console or run the exec command from your CLI.

   For example:

   oc exec -ti my-cluster-zookeeper-0 -- bin/zookeeper-shell.sh localhost:12181 ls /

   Copy to Clipboard Toggle word wrap

   Be sure to use localhost:12181.

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

   Copy to Clipboard Toggle word wrap

2. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

   Output shows the deployment name and readiness

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

   Copy to Clipboard Toggle word wrap

   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. Configure the KafkaConnect custom resource by specifying the container registry in .spec.build.output, and additional connectors in .spec.build.plugins:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
   spec: 

   1

   
     #...
     build:
       output: 

   2

   
         type: docker
         image: my-registry.io/my-org/my-connect-cluster:latest
         pushSecret: my-registry-credentials
       plugins: 

   3

   
         - name: connector-1
           artifacts:
             - type: tgz
               url: <url_to_download_connector_1_artifact>
               sha512sum: <SHA-512_checksum_of_connector_1_artifact>
         - name: connector-2
           artifacts:
             - type: jar
               url: <url_to_download_connector_2_artifact>
               sha512sum: <SHA-512_checksum_of_connector_2_artifact>
     #...

   Copy to Clipboard Toggle word wrap

   1

   The specification for the Kafka Connect cluster.

   2

   (Required) Configuration of the container registry where new images are pushed.

   3

   (Required) List of connector plugins and their artifacts to add to the new container image. Each plugin must be configured with at least one artifact.

2. Create or update the resource:

   $ oc apply -f <kafka_connect_configuration_file>

   Copy to Clipboard Toggle word wrap
3. Wait for the new container image to build, and for the Kafka Connect cluster to be deployed.
4. Use the Kafka Connect REST API or KafkaConnector custom resources to use the connector plugins you added.

Procedure

1. Create a new Dockerfile using registry.redhat.io/amq-streams/kafka-39-rhel9:2.9.3 as the base image:

   FROM registry.redhat.io/amq-streams/kafka-39-rhel9:2.9.3
   USER root:root
   COPY ./my-plugins/ /opt/kafka/plugins/
   USER 1001

   Copy to Clipboard Toggle word wrap

   Example plugins file

   $ tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   │   ├── bson-<version>.jar
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mongodb-<version>.jar
   │   ├── debezium-core-<version>.jar
   │   ├── LICENSE.txt
   │   ├── mongodb-driver-core-<version>.jar
   │   ├── README.md
   │   └── # ...
   ├── debezium-connector-mysql
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mysql-<version>.jar
   │   ├── debezium-core-<version>.jar
   │   ├── LICENSE.txt
   │   ├── mysql-binlog-connector-java-<version>.jar
   │   ├── mysql-connector-java-<version>.jar
   │   ├── README.md
   │   └── # ...
   └── debezium-connector-postgres
       ├── CHANGELOG.md
       ├── CONTRIBUTE.md
       ├── COPYRIGHT.txt
       ├── debezium-connector-postgres-<version>.jar
       ├── debezium-core-<version>.jar
       ├── LICENSE.txt
       ├── postgresql-<version>.jar
       ├── protobuf-java-<version>.jar
       ├── README.md
       └── # ...

   Copy to Clipboard Toggle word wrap

   The COPY command points to the plugin files to copy to the container image.

   This example adds plugins for Debezium connectors (MongoDB, MySQL, and PostgreSQL), though not all files are listed for brevity. Debezium running in Kafka Connect looks the same as any other Kafka Connect task.

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

4. Point to the new container image.

   You can point to the image in one of the following ways:

   • Edit the KafkaConnect.spec.image property of the KafkaConnect custom resource.

     If set, this property overrides the STRIMZI_KAFKA_CONNECT_IMAGES environment variable in the Cluster Operator.

     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaConnect
     metadata:
       name: my-connect-cluster
     spec: 

     1

     
       #...
       image: my-new-container-image 

     2

     
       config: 

     3

     
         #...

     Copy to Clipboard Toggle word wrap

     1

     The specification for the Kafka Connect cluster.

     2

     The docker image for Kafka Connect pods.

     3

     Configuration of the Kafka Connect workers (not connectors).

   • Edit the STRIMZI_KAFKA_CONNECT_IMAGES environment variable in the install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml file to point to the new container image, and then reinstall the Cluster Operator.

Procedure

1. Add the FileStreamSourceConnector and FileStreamSinkConnector plugins to Kafka Connect in one of the following ways:

   • Configure Kafka Connect to build a new container image with plugins automatically
   • Create a Docker image from the base Kafka Connect image (manually or using continuous integration)

2. Set the strimzi.io/use-connector-resources annotation to true in the Kafka Connect configuration.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: "true"
   spec:
       # ...

   Copy to Clipboard Toggle word wrap

   With the KafkaConnector resources enabled, the Cluster Operator watches for them.

3. Edit the examples/connect/source-connector.yaml file:

   Example source connector configuration

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector  

   1

   
     labels:
       strimzi.io/cluster: my-connect-cluster 

   2

   
   spec:
     class: org.apache.kafka.connect.file.FileStreamSourceConnector 

   3

   
     tasksMax: 2 

   4

   
     autoRestart: 

   5

   
       enabled: true
     config: 

   6

   
       file: "/opt/kafka/LICENSE" 

   7

   
       topic: my-topic 

   8

   
       # ...

   Copy to Clipboard Toggle word wrap

   1

   Name of the KafkaConnector resource, which is used as the name of the connector. Use any name that is valid for an OpenShift resource.

   2

   Name of the Kafka Connect cluster to create the connector instance in. Connectors must be deployed to the same namespace as the Kafka Connect cluster they link to.

   3

   Full name of the connector class. This should be present in the image being used by the Kafka Connect cluster.

   4

   Maximum number of Kafka Connect tasks that the connector can create.

   5

   Enables automatic restarts of failed connectors and tasks. By default, the number of restarts is indefinite, but you can set a maximum on the number of automatic restarts using the maxRestarts property.

   6

   Connector configuration as key-value pairs.

   7

   Location of the external data file. In this example, we’re configuring the FileStreamSourceConnector to read from the /opt/kafka/LICENSE file.

   8

   Kafka topic to publish the source data to.

4. Create the source KafkaConnector in your OpenShift cluster:

   oc apply -f examples/connect/source-connector.yaml

   Copy to Clipboard Toggle word wrap

5. Create an examples/connect/sink-connector.yaml file:

   touch examples/connect/sink-connector.yaml

   Copy to Clipboard Toggle word wrap

6. Paste the following YAML into the sink-connector.yaml file:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-sink-connector
     labels:
       strimzi.io/cluster: my-connect
   spec:
     class: org.apache.kafka.connect.file.FileStreamSinkConnector 

   1

   
     tasksMax: 2
     config: 

   2

   
       file: "/tmp/my-file" 

   3

   
       topics: my-topic 

   4

   Copy to Clipboard Toggle word wrap

   1

   Full name or alias of the connector class. This should be present in the image being used by the Kafka Connect cluster.

   2

   Connector configuration as key-value pairs.

   3

   Temporary file to publish the source data to.

   4

   Kafka topic to read the source data from.

7. Create the sink KafkaConnector in your OpenShift cluster:

   oc apply -f examples/connect/sink-connector.yaml

   Copy to Clipboard Toggle word wrap

8. Check that the connector resources were created:

   oc get kctr --selector strimzi.io/cluster=<my_connect_cluster> -o name
   
   my-source-connector
   my-sink-connector

   Copy to Clipboard Toggle word wrap

   Replace <my_connect_cluster> with the name of your Kafka Connect cluster.

9. In the container, execute kafka-console-consumer.sh to read the messages that were written to the topic by the source connector:

   oc exec <my_kafka_cluster>-kafka-0 -i -t -- bin/kafka-console-consumer.sh --bootstrap-server <my_kafka_cluster>-kafka-bootstrap.NAMESPACE.svc:9092 --topic my-topic --from-beginning

   Copy to Clipboard Toggle word wrap

   Replace <my_kafka_cluster> with the name of your Kafka cluster.

1. Deploy KafkaConnector resources with the configuration to create your connector instances.
2. Enable KafkaConnector resources in your Kafka Connect configuration by setting the strimzi.io/use-connector-resources annotation to true.

Procedure

1. Deploy Kafka MirrorMaker to your OpenShift cluster:

   For MirrorMaker:

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

   Copy to Clipboard Toggle word wrap

   For MirrorMaker 2:

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

   Copy to Clipboard Toggle word wrap

2. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

2. Check the status of the deployment:

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

   Output shows the deployment name and readiness

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

   Copy to Clipboard Toggle word wrap

   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>

   Copy to Clipboard Toggle word wrap

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

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

   Copy to Clipboard Toggle word wrap
   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_FULL_RECONCILIATION_INTERVAL_MS 

   4

   
                 value: "120000"
               - name: STRIMZI_LOG_LEVEL 

   5

   
                 value: INFO
               - name: STRIMZI_TLS_ENABLED 

   6

   
                 value: "false"
               - name: STRIMZI_JAVA_OPTS 

   7

   
                 value: "-Xmx=512M -Xms=256M"
               - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 

   8

   
                 value: "-Djavax.net.debug=verbose -DpropertyName=value"
               - name: STRIMZI_PUBLIC_CA 

   9

   
                 value: "false"
               - name: STRIMZI_TLS_AUTH_ENABLED 

   10

   
                 value: "false"
               - name: STRIMZI_SASL_ENABLED 

   11

   
                 value: "false"
               - name: STRIMZI_SASL_USERNAME 

   12

   
                 value: "admin"
               - name: STRIMZI_SASL_PASSWORD 

   13

   
                 value: "password"
               - name: STRIMZI_SASL_MECHANISM 

   14

   
                 value: "scram-sha-512"
               - name: STRIMZI_SECURITY_PROTOCOL 

   15

   
                 value: "SSL"
               - name: STRIMZI_USE_FINALIZERS
                 value: "false" 

   16

   Copy to Clipboard Toggle word wrap

   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

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

   5

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

   6

   Enables TLS support for encrypted communication with the Kafka brokers.

   7

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

   8

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

   9

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

   10

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

   11

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

   12

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

   13

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

   14

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

   15

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

   16

   Set STRIMZI_USE_FINALIZERS to false if you do not want to use finalizers to control topic deletion.

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

   Copy to Clipboard Toggle word wrap

   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. If you need to configure custom SASL authentication, you can define the necessary authentication properties using the STRIMZI_SASL_CUSTOM_CONFIG_JSON environment variable for the standalone operator. For example, this configuration may be used for accessing a Kafka cluster in a cloud provider with a custom login module like the Amazon MSK Library for AWS Identity and Access Management (aws-msk_iam-auth).

   The property STRIMZI_ALTERABLE_TOPIC_CONFIG defaults to ALL, allowing all .spec.config properties to be set in the KafkaTopic resource. If this setting is not suitable for a managed Kafka service, do as follows:

   • If only a subset of properties is configurable, list them as comma-separated values.
   • If no properties are to be configured, use NONE, which is equivalent to an empty property list.
   Note

   Only Kafka configuration properties starting with sasl. can be set with the STRIMZI_SASL_CUSTOM_CONFIG_JSON environment variable.

   Example custom SASL configuration

   # ....
   env:
     - name: STRIMZI_SASL_ENABLED
       value: "true"
     - name: STRIMZI_SECURITY_PROTOCOL
       value: SASL_SSL
     - name: STRIMZI_SKIP_CLUSTER_CONFIG_REVIEW 

   1

   
       value: "true"
     - name: STRIMZI_ALTERABLE_TOPIC_CONFIG 

   2

   
       value: compression.type, max.message.bytes, message.timestamp.difference.max.ms, message.timestamp.type, retention.bytes, retention.ms
     - name: STRIMZI_SASL_CUSTOM_CONFIG_JSON 

   3

   
       value: |
         {
           "sasl.mechanism": "AWS_MSK_IAM",
           "sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
           "sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler"
         }
     - name: STRIMZI_PUBLIC_CA
       value: "true"
     - name: STRIMZI_TRUSTSTORE_LOCATION
       value: /etc/pki/java/cacerts
     - name: STRIMZI_TRUSTSTORE_PASSWORD
       value: changeit
     - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS
       value: my-kafka-cluster-.kafka-serverless.us-east-1.amazonaws.com:9098
   # ...

   Copy to Clipboard Toggle word wrap

   1

   Disables cluster configuration lookup for managed Kafka services that don’t allow topic configuration changes.

   2

   Defines the topic configuration properties that can be updated based on the limitations set by managed Kafka services.

   3

   Specifies the SASL properties to be set in JSON format. Only properties starting with sasl. are allowed.

   Example Dockerfile with external jars

   FROM registry.redhat.io/amq-streams/strimzi-rhel9-operator:2.9.3
   
   USER root
   
   RUN mkdir -p ${STRIMZI_HOME}/external-libs
   RUN chmod +rx ${STRIMZI_HOME}/external-libs
   
   COPY ./aws-msk-iam-auth-and-dependencies/* ${STRIMZI_HOME}/external-libs/
   ENV JAVA_CLASSPATH=${STRIMZI_HOME}/external-libs/*
   
   USER 1001

   Copy to Clipboard Toggle word wrap

5. Apply the changes to the Deployment configuration to deploy the Topic Operator.

6. Check the status of the deployment:

   oc get deployments

   Copy to Clipboard Toggle word wrap

   Output shows the deployment name and readiness

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

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

4. Check the status of the deployment:

   oc get deployments

   Copy to Clipboard Toggle word wrap

   Output shows the deployment name and readiness

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

   Copy to Clipboard Toggle word wrap

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