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/amq7/amq-streams-rhel8-operator:2.3.0
           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/amq7/amq-streams-rhel8-operator:2.3.0
           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.

   With the default deployment, you install 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 shows 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: {}

2. Configure the Topic Operator spec using the properties described in 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 shows 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 shows 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 deployments -n <my_cluster_operator_namespace>

   Output shows the deployment name and readiness

   NAME                        READY  UP-TO-DATE  AVAILABLE
   my-connect-cluster-connect  1/1    1           1

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

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

Procedure

1. Deploy Kafka MirrorMaker to your OpenShift cluster:

   For MirrorMaker:

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

   For MirrorMaker 2.0:

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

2. 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
   my-mirror-maker-mirror-maker  1/1    1           1
   my-mm2-cluster-mirrormaker2   1/1    1           1

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

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

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 deployments -n <my_cluster_operator_namespace>

   Output shows the deployment name and readiness

   NAME              READY  UP-TO-DATE  AVAILABLE
   my-bridge-bridge  1/1    1           1

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

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

Procedure

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

   oc get pods -o name
   
   pod/kafka-consumer
   # ...
   pod/my-bridge-bridge-7cbd55496b-nclrt

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

   oc port-forward pod/my-bridge-bridge-7cbd55496b-nclrt 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

   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

   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_LOG_LEVEL 9
                 value: INFO
               - name: STRIMZI_GC_LOG_ENABLED 10
                 value: "true"
               - name: STRIMZI_CA_VALIDITY 11
                 value: "365"
               - name: STRIMZI_CA_RENEWAL 12
                 value: "30"
               - name: STRIMZI_JAVA_OPTS 13
                 value: "-Xmx=512M -Xms=256M"
               - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 14
                 value: "-Djavax.net.debug=verbose -DpropertyName=value"
               - name: STRIMZI_SECRET_PREFIX 15
                 value: "kafka-"
               - name: STRIMZI_ACLS_ADMIN_API_SUPPORTED 16
                 value: "true"
               - name: STRIMZI_MAINTENANCE_TIME_WINDOWS 17
                 value: '* * 8-10 * * ?;* * 14-15 * * ?'
               - name: STRIMZI_KAFKA_ADMIN_CLIENT_CONFIGURATION 18
                 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 Certificate Authority that signs new user certificates for mTLS authentication.

   4

   The OpenShift Secret that contains the private key (ca.key) value of the Certificate Authority 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 level for printing logging messages. You can set the level to ERROR, WARNING, INFO, DEBUG, or TRACE.

   10

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

   11

   The validity period for the Certificate Authority. The default is 365 days.

   12

   The renewal period for the Certificate Authority. 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.

   13

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

   14

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

   15

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

   16

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

   17

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

   18

   (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 keystore (entity-operator.p12) with the private key and certificate for mTLS authentication against the Kafka cluster. The Secret must also contain the password (entity-operator.password) for accessing the keystore.

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. Check that monitoring for user-defined projects is enabled:

   oc get pods -n openshift-user-workload-monitoring

   If enabled, pods for the monitoring components are returned. For example:

   NAME                                   READY   STATUS    RESTARTS   AGE
   prometheus-operator-5cc59f9bc6-kgcq8   1/1     Running   0          25s
   prometheus-user-workload-0             5/5     Running   1          14s
   prometheus-user-workload-1             5/5     Running   1          14s
   thanos-ruler-user-workload-0           3/3     Running   0          14s
   thanos-ruler-user-workload-1           3/3     Running   0          14s

   If no pods are returned, monitoring for user-defined projects is disabled. See the Prerequisites in Section 8.5, “Viewing Kafka metrics and dashboards in OpenShift”.

2. Multiple PodMonitor resources are defined in examples/metrics/prometheus-install/strimzi-pod-monitor.yaml.

   For each PodMonitor resource, edit the spec.namespaceSelector.matchNames property:

   apiVersion: monitoring.coreos.com/v1
   kind: PodMonitor
   metadata:
     name: cluster-operator-metrics
     labels:
       app: strimzi
   spec:
     selector:
       matchLabels:
         strimzi.io/kind: cluster-operator
     namespaceSelector:
       matchNames:
         - <project-name> 1
     podMetricsEndpoints:
     - path: /metrics
       port: http
   # ...

   1

   The project where the pods to scrape the metrics from are running, for example, Kafka.

3. Deploy the strimzi-pod-monitor.yaml file to the project where your Kafka cluster is running:

   oc apply -f strimzi-pod-monitor.yaml -n MY-PROJECT

4. Deploy the example Prometheus rules to the same project:

   oc apply -f prometheus-rules.yaml -n MY-PROJECT

Procedure

1. Create a ServiceAccount for Grafana. Here the resource is named grafana-serviceaccount.

   apiVersion: v1
   kind: ServiceAccount
   metadata:
     name: grafana-serviceaccount
     labels:
       app: strimzi

2. Deploy the ServiceAccount to the project containing your Kafka cluster:

   oc apply -f GRAFANA-SERVICEACCOUNT -n MY-PROJECT

3. Create a ClusterRoleBinding resource that assigns the cluster-monitoring-view role to the Grafana ServiceAccount. Here the resource is named grafana-cluster-monitoring-binding.

   apiVersion: rbac.authorization.k8s.io/v1
   kind: ClusterRoleBinding
   metadata:
     name: grafana-cluster-monitoring-binding
     labels:
       app: strimzi
   subjects:
     - kind: ServiceAccount
       name: grafana-serviceaccount
       namespace: <my-project> 1
   roleRef:
     kind: ClusterRole
     name: cluster-monitoring-view
     apiGroup: rbac.authorization.k8s.io

   1

   Name of your project.

4. Deploy the ClusterRoleBinding to the project containing your Kafka cluster:

   oc apply -f <grafana-cluster-monitoring-binding> -n <my-project>

Procedure

1. Get the access token of the Grafana ServiceAccount:

   oc serviceaccounts get-token grafana-serviceaccount -n MY-PROJECT

   Copy the access token to use in the next step.

2. Create a datasource.yaml file containing the Thanos Querier configuration for Grafana.

   Paste the access token into the httpHeaderValue1 property as indicated.

   apiVersion: 1
   
   datasources:
   - name: Prometheus
     type: prometheus
     url: https://thanos-querier.openshift-monitoring.svc.cluster.local:9091
     access: proxy
     basicAuth: false
     withCredentials: false
     isDefault: true
     jsonData:
       timeInterval: 5s
       tlsSkipVerify: true
       httpHeaderName1: "Authorization"
     secureJsonData:
       httpHeaderValue1: "Bearer ${GRAFANA-ACCESS-TOKEN}" 1
     editable: true

   1

   GRAFANA-ACCESS-TOKEN: The value of the access token for the Grafana ServiceAccount.

3. Create a config map named grafana-config from the datasource.yaml file:

   oc create configmap grafana-config --from-file=datasource.yaml -n MY-PROJECT

4. Create a Grafana application consisting of a Deployment and a Service.

   The grafana-config config map is mounted as a volume for the datasource configuration.

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: grafana
     labels:
       app: strimzi
   spec:
     replicas: 1
     selector:
       matchLabels:
         name: grafana
     template:
       metadata:
         labels:
           name: grafana
       spec:
         serviceAccountName: grafana-serviceaccount
         containers:
         - name: grafana
           image: grafana/grafana:9.3.1
           ports:
           - name: grafana
             containerPort: 3000
             protocol: TCP
           volumeMounts:
           - name: grafana-data
             mountPath: /var/lib/grafana
           - name: grafana-logs
             mountPath: /var/log/grafana
           - name: grafana-config
             mountPath: /etc/grafana/provisioning/datasources/datasource.yaml
             readOnly: true
             subPath: datasource.yaml
           readinessProbe:
             httpGet:
               path: /api/health
               port: 3000
             initialDelaySeconds: 5
             periodSeconds: 10
           livenessProbe:
             httpGet:
               path: /api/health
               port: 3000
             initialDelaySeconds: 15
             periodSeconds: 20
         volumes:
         - name: grafana-data
           emptyDir: {}
         - name: grafana-logs
           emptyDir: {}
         - name: grafana-config
           configMap:
             name: grafana-config
   ---
   apiVersion: v1
   kind: Service
   metadata:
     name: grafana
     labels:
       app: strimzi
   spec:
     ports:
     - name: grafana
       port: 3000
       targetPort: 3000
       protocol: TCP
     selector:
       name: grafana
     type: ClusterIP

5. Deploy the Grafana application to the project containing your Kafka cluster:

   oc apply -f <grafana-application> -n <my-project>

Procedure

1. Get the details of the Route to the Grafana Service. For example:

   oc get routes
   
   NAME               HOST/PORT                         PATH  SERVICES
   MY-GRAFANA-ROUTE   MY-GRAFANA-ROUTE-amq-streams.net        grafana

2. In a web browser, access the Grafana login screen using the URL for the Route host and port.

3. Enter your user name and password, and then click Log In.

   The default Grafana user name and password are both admin. After logging in for the first time, you can change the password.

4. In Configuration > Data Sources, check that the Prometheus data source was created. The data source was created in Section 8.5.5, “Deploying Grafana with a Prometheus datasource”.
5. Click the + icon and then click Import.
6. In examples/metrics/grafana-dashboards, copy the JSON of the dashboard to import.
7. Paste the JSON into the text box, and then click Load.
8. Repeat steps 5-7 for the other example Grafana dashboards.

1. In the spec.template property, configure the tracer service.

   • Use the tracing environment variables as template configuration properties.
   • For OpenTelemetry, set the spec.tracing.type property to opentelemetry.
   • For OpenTracing, set the spec.tracing.type property to jaeger.

   Example tracing configuration for Kafka Connect using OpenTelemetry

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
   spec:
     #...
     template:
       connectContainer:
         env:
           - name: OTEL_SERVICE_NAME
             value: my-otel-service
           - name: OTEL_EXPORTER_JAEGER_ENDPOINT
             value: "http://jaeger-host:14250"
     tracing:
       type: opentelemetry
     #...

   Example tracing configuration for MirrorMaker using OpenTelemetry

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker
   metadata:
     name: my-mirror-maker
   spec:
     #...
     template:
       mirrorMakerContainer:
         env:
           - name: OTEL_SERVICE_NAME
             value: my-otel-service
           - name: OTEL_EXPORTER_JAEGER_ENDPOINT
             value: "http://jaeger-host:14250"
     tracing:
       type: opentelemetry
   #...

   Example tracing configuration for MirrorMaker 2.0 using OpenTelemetry

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mm2-cluster
   spec:
     #...
     template:
       connectContainer:
         env:
           - name: OTEL_SERVICE_NAME
             value: my-otel-service
           - name: OTEL_EXPORTER_JAEGER_ENDPOINT
             value: "http://jaeger-host:14250"
     tracing:
       type: opentelemetry
   #...

   Example tracing configuration for the Kafka Bridge using OpenTelemetry

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     #...
     template:
       bridgeContainer:
         env:
           - name: OTEL_SERVICE_NAME
             value: my-otel-service
           - name: OTEL_EXPORTER_JAEGER_ENDPOINT
             value: "http://jaeger-host:14250"
     tracing:
       type: opentelemetry
   #...

   Example tracing configuration for Kafka Connect using OpenTracing

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
   spec:
     #...
     template:
       connectContainer:
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing:
       type: jaeger
     #...

   Example tracing configuration for MirrorMaker using OpenTracing

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker
   metadata:
     name: my-mirror-maker
   spec:
     #...
     template:
       mirrorMakerContainer:
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing:
       type: jaeger
   #...

   Example tracing configuration for MirrorMaker 2.0 using OpenTracing

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mm2-cluster
   spec:
     #...
     template:
       connectContainer:
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing:
       type: jaeger
   #...

   Example tracing configuration for the Kafka Bridge using OpenTracing

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     #...
     template:
       bridgeContainer:
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing:
       type: jaeger
   #...

2. Create or update the resource:

   oc apply -f <resource_configuration_file>

1. Add the Maven dependencies to the pom.xml file for the client application:

   Dependencies for OpenTelemetry

   <dependency>
       <groupId>io.opentelemetry</groupId>
       <artifactId>opentelemetry-sdk-extension-autoconfigure</artifactId>
       <version>1.18.0-alpha</version>
   </dependency>
   <dependency>
     <groupId>io.opentelemetry.instrumentation</groupId>
     <artifactId>opentelemetry-kafka-clients-{OpenTelemetryKafkaClient}</artifactId>
     <version>1.18.0-alpha</version>
   </dependency>
   <dependency>
     <groupId>io.opentelemetry</groupId>
     <artifactId>opentelemetry-exporter-jaeger</artifactId>
     <version>1.18.0</version>
   </dependency>

   Dependencies for OpenTracing

   <dependency>
       <groupId>io.jaegertracing</groupId>
       <artifactId>jaeger-client</artifactId>
       <version>1.3.2</version>
   </dependency>
   <dependency>
     <groupId>io.opentracing.contrib</groupId>
     <artifactId>opentracing-kafka-client</artifactId>
     <version>0.1.15</version>
   </dependency>

2. Define the configuration of the tracer using the tracing environment variables.

3. Create a tracer, which is initialized with the environment variables:

   Creating a tracer for OpenTelemetry

   OpenTelemetry ot = GlobalOpenTelemetry.get();

   Creating a tracer for OpenTracing

   Tracer tracer = Configuration.fromEnv().getTracer();

4. Register the tracer as a global tracer:

   GlobalTracer.register(tracer);

5. Instrument your client:

   • Section 9.3.4, “Instrumenting producers and consumers for tracing”
   • Section 9.3.5, “Instrumenting Kafka Streams applications for tracing”

Procedure

1. Add the tracing artifacts to the /opt/kafka/libs/ directory of the AMQ Streams Kafka image.

   You can use the Kafka container image on the Red Hat Ecosystem Catalog as a base image for creating a new custom image.

   OpenTelemetry artifact for Zipkin

   io.opentelemetry:opentelemetry-exporter-zipkin

2. Set the tracing exporter and endpoint for the new tracing implementation.

   Example Zikpin tracer configuration

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mm2-cluster
   spec:
     #...
     template:
       connectContainer:
         env:
           - name: OTEL_SERVICE_NAME
             value: my-zipkin-service
           - name: OTEL_EXPORTER_ZIPKIN_ENDPOINT
             value: http://zipkin-exporter-host-name:9411/api/v2/spans 1
           - name: OTEL_TRACES_EXPORTER
             value: zipkin 2
     tracing:
       type: opentelemetry
   #...

   1

   Specifies the Zipkin endpoint to connect to.

   2

   The Zipkin exporter.

1. Upgrade AMQ Streams to version 1.7 following the standard sequence.
2. Convert AMQ Streams custom resources to v1beta2 using the Red Hat AMQ Streams API Conversion Tool provided with AMQ Streams 1.8.

3. Do one of the following:

   • Upgrade to AMQ Streams 1.8 (where the ControlPlaneListener feature gate is disabled by default).
   • Upgrade to AMQ Streams 2.0 or 2.2 (where the ControlPlaneListener feature gate is enabled by default) with the ControlPlaneListener feature gate disabled.
4. Enable the ControlPlaneListener feature gate.
5. Upgrade to AMQ Streams 2.3 following the standard sequence.

1. Upgrade to AMQ Streams 1.7.
2. Download the Red Hat AMQ Streams API Conversion Tool provided with AMQ Streams 1.8.

3. Convert custom resources and CRDs to v1beta2.

   For more information, see the AMQ Streams 1.7 upgrade documentation.

4. In the OperatorHub, delete version 1.7 of the Red Hat Integration - AMQ Streams Operator.

5. If it also exists, delete version 2.3 of the Red Hat Integration - AMQ Streams Operator.

   If it does not exist, go to the next step.

   If the Approval Strategy for the AMQ Streams Operator was set to Automatic, version 2.3 of the operator might already exist in your cluster. If you did not convert custom resources and CRDs to the v1beta2 API version before release, the operator-managed custom resources and CRDs will be using the old API version. As a result, the 2.3 Operator is stuck in Pending status. In this situation, you need to delete version 2.3 of the Red Hat Integration - AMQ Streams Operator as well as version 1.7.

   If you delete both operators, reconciliations are paused until the new operator version is installed. Follow the next steps immediately so that any changes to custom resources are not delayed.

6. In the OperatorHub, do one of the following:

   • Upgrade to version 1.8 of the Red Hat Integration - AMQ Streams Operator (where the ControlPlaneListener feature gate is disabled by default).
   • Upgrade to version 2.0 or 2.2 of the Red Hat Integration - AMQ Streams Operator (where the ControlPlaneListener feature gate is enabled by default) with the ControlPlaneListener feature gate disabled.

7. Upgrade to version 2.3 of the Red Hat Integration - AMQ Streams Operator immediately.

   The installed 2.3 operator begins to watch the cluster and performs rolling updates. You might notice a temporary decrease in cluster performance during this process.

Procedure

1. Take note of any configuration changes made to the existing Cluster Operator resources (in the /install/cluster-operator directory). Any changes will be overwritten by the new version of the Cluster Operator.
2. Update your custom resources to reflect the supported configuration options available for AMQ Streams version 2.3.

3. Update the Cluster Operator.

   1. Modify the installation files for the new Cluster Operator version according to the namespace the Cluster Operator is running in.

      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. If you modified one or more environment variables in your existing Cluster Operator Deployment, edit the install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml file to use those environment variables.

4. When you have an updated configuration, deploy it along with the rest of the installation resources:

   oc replace -f install/cluster-operator

   Wait for the rolling updates to complete.

5. If the new Operator version no longer supports the Kafka version you are upgrading from, the Cluster Operator returns an error message to say the version is not supported. Otherwise, no error message is returned.

   • If the error message is returned, upgrade to a Kafka version that is supported by the new Cluster Operator version:

     1. Edit the Kafka custom resource.
     2. Change the spec.kafka.version property to a supported Kafka version.
   • If the error message is not returned, go to the next step. You will upgrade the Kafka version later.

6. Get the image for the Kafka pod to ensure the upgrade was successful:

   oc get pods my-cluster-kafka-0 -o jsonpath='{.spec.containers[0].image}'

   The image tag shows the new Operator version. For example:

   registry.redhat.io/amq7/amq-streams-kafka-33-rhel8:2.3.0

Broker-level consumers first strategy

1. Upgrade all the consuming applications.
2. Change the broker-level log.message.format.version to the new version.
3. Upgrade all the producing applications.

1. Upgrade all the consuming applications.
2. Change the topic-level message.format.version to the new version.
3. Upgrade all the producing applications.

1. Change the topic-level message.format.version to the old version (or rely on the topic defaulting to the broker-level log.message.format.version).
2. Upgrade all the consuming and producing applications.
3. Verify that the upgraded applications function correctly.
4. Change the topic-level message.format.version to the new version.

Procedure

1. Update the Kafka cluster configuration:

   oc edit kafka <my_cluster>

2. If configured, check that the inter.broker.protocol.version and log.message.format.version properties are set to the current version.

   For example, the current version is 3.2 if upgrading from Kafka version 3.2.3 to 3.3.1:

   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.2.3
       config:
         log.message.format.version: "3.2"
         inter.broker.protocol.version: "3.2"
         # ...

   If log.message.format.version and inter.broker.protocol.version are not configured, AMQ Streams automatically updates these versions to the current defaults after the update to the Kafka version in the next step.

   Note

   The value of log.message.format.version and inter.broker.protocol.version must be strings to prevent them from being interpreted as floating point numbers.

3. Change the Kafka.spec.kafka.version to specify the new Kafka version; leave the log.message.format.version and inter.broker.protocol.version at the defaults for the current Kafka version.

   Note

   Changing the kafka.version ensures that all brokers in the cluster will be upgraded to start using the new broker binaries. During this process, some brokers are using the old binaries while others have already upgraded to the new ones. Leaving the inter.broker.protocol.version unchanged at the current setting ensures that the brokers can continue to communicate with each other throughout the upgrade.

   For example, if upgrading from Kafka 3.2.3 to 3.3.1:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.3.1 1
       config:
         log.message.format.version: "3.2" 2
         inter.broker.protocol.version: "3.2" 3
         # ...

   1

   Kafka version is changed to the new version.

   2

   Message format version is unchanged.

   3

   Inter-broker protocol version is unchanged.

   Warning

   You cannot downgrade Kafka if the inter.broker.protocol.version for the new Kafka version changes. The inter-broker protocol version determines the schemas used for persistent metadata stored by the broker, including messages written to __consumer_offsets. The downgraded cluster will not understand the messages.

4. If the image for the Kafka cluster is defined in the Kafka custom resource, in Kafka.spec.kafka.image, update the image to point to a container image with the new Kafka version.

   See Kafka version and image mappings

5. Save and exit the editor, then wait for rolling updates to complete.

   Check the progress of the rolling updates by watching the pod state transitions:

   oc get pods my-cluster-kafka-0 -o jsonpath='{.spec.containers[0].image}'

   The rolling updates ensure that each pod is using the broker binaries for the new version of Kafka.

6. Depending on your chosen strategy for upgrading clients, upgrade all client applications to use the new version of the client binaries.

   If required, set the version property for Kafka Connect and MirrorMaker as the new version of Kafka:

   1. For Kafka Connect, update KafkaConnect.spec.version.
   2. For MirrorMaker, update KafkaMirrorMaker.spec.version.
   3. For MirrorMaker 2.0, update KafkaMirrorMaker2.spec.version.

7. If configured, update the Kafka resource to use the new inter.broker.protocol.version version. Otherwise, go to step 9.

   For example, if upgrading to Kafka 3.3.1:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.3.1
       config:
         log.message.format.version: "3.2"
         inter.broker.protocol.version: "3.3"
         # ...

8. Wait for the Cluster Operator to update the cluster.

9. If configured, update the Kafka resource to use the new log.message.format.version version. Otherwise, go to step 10.

   For example, if upgrading to Kafka 3.3.1:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.3.1
       config:
         log.message.format.version: "3.3"
         inter.broker.protocol.version: "3.3"
         # ...

   Important

   From Kafka 3.0.0, when the inter.broker.protocol.version is set to 3.0 or higher, the log.message.format.version option is ignored and doesn’t need to be set.

10. Wait for the Cluster Operator to update the cluster.

    • The Kafka cluster and clients are now using the new Kafka version.
    • The brokers are configured to send messages using the inter-broker protocol version and message format version of the new version of Kafka.

Procedure

1. Update the Kafka cluster configuration.

   oc edit kafka KAFKA-CONFIGURATION-FILE

2. Change the Kafka.spec.kafka.version to specify the previous version.

   For example, if downgrading from Kafka 3.3.1 to 3.2.3:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.2.3 1
       config:
         log.message.format.version: "3.2" 2
         inter.broker.protocol.version: "3.2" 3
         # ...

   1

   Kafka version is changed to the previous version.

   2

   Message format version is unchanged.

   3

   Inter-broker protocol version is unchanged.

   Note

   The value of log.message.format.version and inter.broker.protocol.version must be strings to prevent them from being interpreted as floating point numbers.

3. If the image for the Kafka version is different from the image defined in STRIMZI_KAFKA_IMAGES for the Cluster Operator, update Kafka.spec.kafka.image.

   See Section 10.5.3, “Kafka version and image mappings”

4. Save and exit the editor, then wait for rolling updates to complete.

   Check the update in the logs or by watching the pod state transitions:

   oc logs -f CLUSTER-OPERATOR-POD-NAME | grep -E "Kafka version downgrade from [0-9.]+ to [0-9.]+, phase ([0-9]+) of \1 completed"

   oc get pod -w

   Check the Cluster Operator logs for an INFO level message:

   Reconciliation #NUM(watch) Kafka(NAMESPACE/NAME): Kafka version downgrade from FROM-VERSION to TO-VERSION, phase 1 of 1 completed

5. Downgrade all client applications (consumers) to use the previous version of the client binaries.

   The Kafka cluster and clients are now using the previous Kafka version.

6. If you are reverting back to a version of AMQ Streams earlier than 1.7, which uses ZooKeeper for the storage of topic metadata, delete the internal topic store topics from the Kafka cluster.

   oc run kafka-admin -ti --image=registry.redhat.io/amq7/amq-streams-kafka-33-rhel8:2.3.0 --rm=true --restart=Never -- ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic __strimzi-topic-operator-kstreams-topic-store-changelog --delete && ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic __strimzi_store_topic --delete