Using AMQ Streams on OpenShift

Procedure

1. Find the name of the StatefulSet that controls the Kafka pods you want to manually update.

   For example, if your Kafka cluster is named my-cluster, the corresponding StatefulSet is named my-cluster-kafka.

2. Annotate the StatefulSet resource in OpenShift. For example, using oc annotate:

   oc annotate statefulset cluster-name-kafka strimzi.io/manual-rolling-update=true

3. Wait for the next reconciliation to occur (every two minutes by default). A rolling update of all pods within the annotated StatefulSet is triggered, as long as the annotation was detected by the reconciliation process. When the rolling update of all the pods is complete, the annotation is removed from the StatefulSet.

Procedure

1. Find the name of the StatefulSet that controls the ZooKeeper pods you want to manually update.

   For example, if your Kafka cluster is named my-cluster, the corresponding StatefulSet is named my-cluster-zookeeper.

2. Annotate the StatefulSet resource in OpenShift. For example, using oc annotate:

   oc annotate statefulset cluster-name-zookeeper strimzi.io/manual-rolling-update=true

3. Wait for the next reconciliation to occur (every two minutes by default). A rolling update of all pods within the annotated StatefulSet is triggered, as long as the annotation was detected by the reconciliation process. When the rolling update of all the pods is complete, the annotation is removed from the StatefulSet.

Procedure

1. Find the name of the Pod that you want to delete.

   For example, if the cluster is named cluster-name, the pods are named cluster-name-kafka-index, where index starts at zero and ends at the total number of replicas.

2. Annotate the Pod resource in OpenShift.

   Use oc annotate:

   oc annotate pod cluster-name-kafka-index strimzi.io/delete-pod-and-pvc=true

3. Wait for the next reconciliation, when the annotated pod with the underlying persistent volume claim will be deleted and then recreated.

Procedure

1. Find the name of the Pod that you want to delete.

   For example, if the cluster is named cluster-name, the pods are named cluster-name-zookeeper-index, where index starts at zero and ends at the total number of replicas.

2. Annotate the Pod resource in OpenShift.

   Use oc annotate:

   oc annotate pod cluster-name-zookeeper-index strimzi.io/delete-pod-and-pvc=true

3. Wait for the next reconciliation, when the annotated pod with the underlying persistent volume claim will be deleted and then recreated.

Procedure

1. Apply the strimzi.io/force-renew annotation to the Secret that contains the CA certificate that you want to renew.

   Table 2.1. Annotation for the Secret that forces renewal of certificates

   Certificate	Secret	Annotate command
   Cluster CA	KAFKA-CLUSTER-NAME-cluster-ca-cert	oc annotate secret KAFKA-CLUSTER-NAME-cluster-ca-cert strimzi.io/force-renew=true
   Clients CA	KAFKA-CLUSTER-NAME-clients-ca-cert	oc annotate secret KAFKA-CLUSTER-NAME-clients-ca-cert strimzi.io/force-renew=true

   At the next reconciliation the Cluster Operator will generate a new CA certificate for the Secret that you annotated. If maintenance time windows are configured, the Cluster Operator will generate the new CA certificate at the first reconciliation within the next maintenance time window.

   Client applications must reload the cluster and clients CA certificates that were renewed by the Cluster Operator.

2. Check the period the CA certificate is valid:

   For example, using an openssl command:

   oc get secret CA-CERTIFICATE-SECRET -o 'jsonpath={.data.CA-CERTIFICATE}' | base64 -d | openssl x509 -subject -issuer -startdate -enddate -noout

   CA-CERTIFICATE-SECRET is the name of the Secret, which is KAFKA-CLUSTER-NAME-cluster-ca-cert for the cluster CA certificate and KAFKA-CLUSTER-NAME-clients-ca-cert for the clients CA certificate.

   CA-CERTIFICATE is the name of the CA certificate, such as jsonpath={.data.ca\.crt}.

   The command returns a notBefore and notAfter date, which is the validity period for the CA certificate.

   For example, for a cluster CA certificate:

   subject=O = io.strimzi, CN = cluster-ca v0
   issuer=O = io.strimzi, CN = cluster-ca v0
   notBefore=Jun 30 09:43:54 2020 GMT
   notAfter=Jun 30 09:43:54 2021 GMT

3. Delete old certificates from the Secret.

   When components are using the new certificates, older certificates might still be active. Delete the old certificates to remove any potential security risk.

Procedure

1. Edit the spec properties for the KafkaConnect or KafkaConnectS2I resource.

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

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect 1
   metadata:
     name: my-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: "true" 2
   spec:
     replicas: 3 3
     authentication: 4
       type: tls
       certificateAndKey:
         certificate: source.crt
         key: source.key
         secretName: my-user-source
     bootstrapServers: my-cluster-kafka-bootstrap:9092 5
     tls: 6
       trustedCertificates:
         - secretName: my-cluster-cluster-cert
           certificate: ca.crt
         - secretName: my-cluster-cluster-cert
           certificate: ca2.crt
     config: 7
       group.id: my-connect-cluster
       offset.storage.topic: my-connect-cluster-offsets
       config.storage.topic: my-connect-cluster-configs
       status.storage.topic: my-connect-cluster-status
       key.converter: org.apache.kafka.connect.json.JsonConverter
       value.converter: org.apache.kafka.connect.json.JsonConverter
       key.converter.schemas.enable: true
       value.converter.schemas.enable: true
       config.storage.replication.factor: 3
       offset.storage.replication.factor: 3
       status.storage.replication.factor: 3
     externalConfiguration: 8
       env:
         - name: AWS_ACCESS_KEY_ID
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsAccessKey
         - name: AWS_SECRET_ACCESS_KEY
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsSecretAccessKey
     resources: 9
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 10
       type: inline
       loggers:
         log4j.rootLogger: "INFO"
     readinessProbe: 11
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     metrics: 12
       lowercaseOutputName: true
       lowercaseOutputLabelNames: true
       rules:
         - pattern: kafka.connect<type=connect-worker-metrics><>([a-z-]+)
           name: kafka_connect_worker_$1
           help: "Kafka Connect JMX metric worker"
           type: GAUGE
         - pattern: kafka.connect<type=connect-worker-rebalance-metrics><>([a-z-]+)
           name: kafka_connect_worker_rebalance_$1
           help: "Kafka Connect JMX metric rebalance information"
           type: GAUGE
     jvmOptions: 13
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 14
     template: 15
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 16
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"

   1

   Use KafkaConnect or KafkaConnectS2I, as required.

   2

   Enables KafkaConnectors for the Kafka Connect cluster.

   3

   The number of replica nodes.

   4

   Authentication for the Kafka Connect cluster, using the TLS mechanism, as shown here, using OAuth bearer tokens, or a SASL-based SCRAM-SHA-512 or PLAIN mechanism. By default, Kafka Connect connects to Kafka brokers using a plain text connection.

   5

   Bootstrap server for connection to the Kafka Connect cluster.

   6

   TLS encryption with key names under which TLS certificates are stored in X.509 format for the cluster. If certificates are stored in the same secret, it can be listed multiple times.

   7

   Kafka Connect configuration of workers (not connectors). Standard Apache Kafka configuration may be provided, restricted to those properties not managed directly by AMQ Streams.

   8

   External configuration for Kafka connectors using environment variables, as shown here, or volumes.

   9

   Requests for reservation of supported resources, currently cpu and memory, and limits to specify the maximum resources that can be consumed.

   10

   Specified Kafka Connect loggers and log levels added directly (inline) or indirectly (external) through a ConfigMap. A custom ConfigMap must be placed under the log4j.properties or log4j2.properties key. For the Kafka Connect log4j.rootLogger logger, you can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.

   11

   Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).

   12

   Prometheus metrics, which are enabled with configuration for the Prometheus JMX exporter in this example. You can enable metrics without further configuration using metrics: {}.

   13

   JVM configuration options to optimize performance for the Virtual Machine (VM) running Kafka Connect.

   14

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

   15

   Template customization. Here a pod is scheduled with anti-affinity, so the pod is not scheduled on nodes with the same hostname.

   16

   Environment variables are also set for distributed tracing using Jaeger.

2. Create or update the resource:

   oc apply -f KAFKA-CONNECT-CONFIG-FILE

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

Procedure

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

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

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

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

2. Create or update the resource.

   oc apply -f KAFKA-USER-CONFIG-FILE

Procedure

1. Edit the spec properties for the KafkaMirrorMaker resource.

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

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaMirrorMaker
   metadata:
     name: my-mirror-maker
   spec:
     replicas: 3 1
     consumer:
       bootstrapServers: my-source-cluster-kafka-bootstrap:9092 2
       groupId: "my-group" 3
       numStreams: 2 4
       offsetCommitInterval: 120000 5
       tls: 6
         trustedCertificates:
         - secretName: my-source-cluster-ca-cert
           certificate: ca.crt
       authentication: 7
         type: tls
         certificateAndKey:
           secretName: my-source-secret
           certificate: public.crt
           key: private.key
       config: 8
         max.poll.records: 100
         receive.buffer.bytes: 32768
         ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 9
         ssl.enabled.protocols: "TLSv1.2"
         ssl.protocol: "TLSv1.2"
         ssl.endpoint.identification.algorithm: HTTPS 10
     producer:
       bootstrapServers: my-target-cluster-kafka-bootstrap:9092
       abortOnSendFailure: false 11
       tls:
         trustedCertificates:
         - secretName: my-target-cluster-ca-cert
           certificate: ca.crt
       authentication:
         type: tls
         certificateAndKey:
           secretName: my-target-secret
           certificate: public.crt
           key: private.key
       config:
         compression.type: gzip
         batch.size: 8192
         ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 12
         ssl.enabled.protocols: "TLSv1.2"
         ssl.protocol: "TLSv1.2"
         ssl.endpoint.identification.algorithm: HTTPS 13
     whitelist: "my-topic|other-topic" 14
     resources: 15
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 16
       type: inline
       loggers:
         mirrormaker.root.logger: "INFO"
     readinessProbe: 17
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     metrics: 18
       lowercaseOutputName: true
       rules:
         - pattern: "kafka.server<type=(.+), name=(.+)PerSec\\w*><>Count"
           name: "kafka_server_$1_$2_total"
         - pattern: "kafka.server<type=(.+), name=(.+)PerSec\\w*,
           topic=(.+)><>Count"
           name: "kafka_server_$1_$2_total"
           labels:
             topic: "$3"
     jvmOptions: 19
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 20
     template: 21
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 22
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing: 23
       type: jaeger

   1

   The number of replica nodes.

   2

   Bootstrap servers for consumer and producer.

   3

   Group ID for the consumer.

   4

   The number of consumer streams.

   5

   The offset auto-commit interval in milliseconds.

   6

   TLS encryption with key names under which TLS certificates are stored in X.509 format for consumer or producer. If certificates are stored in the same secret, it can be listed multiple times.

   7

   Authentication for consumer or producer, using the TLS mechanism, as shown here, using OAuth bearer tokens, or a SASL-based SCRAM-SHA-512 or PLAIN mechanism.

   8

   Kafka configuration options for consumer and producer.

   9

   SSL properties for external listeners to run with a specific cipher suite for a TLS version.

   10

   Hostname verification is enabled by setting to HTTPS. An empty string disables the verification.

   11

   If the abortOnSendFailure property is set to true, Kafka MirrorMaker will exit and the container will restart following a send failure for a message.

   12

   SSL properties for external listeners to run with a specific cipher suite for a TLS version.

   13

   Hostname verification is enabled by setting to HTTPS. An empty string disables the verification.

   14

   A whitelist of topics mirrored from source to target Kafka cluster.

   15

   Requests for reservation of supported resources, currently cpu and memory, and limits to specify the maximum resources that can be consumed.

   16

   Specified loggers and log levels added directly (inline) or indirectly (external) through a ConfigMap. A custom ConfigMap must be placed under the log4j.properties or log4j2.properties key. MirrorMaker has a single logger called mirrormaker.root.logger. You can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.

   17

   Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).

   18

   Prometheus metrics, which are enabled with configuration for the Prometheus JMX exporter in this example. You can enable metrics without further configuration using metrics: {}.

   19

   JVM configuration options to optimize performance for the Virtual Machine (VM) running Kafka MirrorMaker.

   20

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

   21

   Template customization. Here a pod is scheduled with anti-affinity, so the pod is not scheduled on nodes with the same hostname.

   22

   Environment variables are also set for distributed tracing using Jaeger.

   23

   Distributed tracing is enabled for Jaeger.

   Warning

   With the abortOnSendFailure property set to false, the producer attempts to send the next message in a topic. The original message might be lost, as there is no attempt to resend a failed message.

2. Create or update the resource:

   oc apply -f <your-file>

Procedure

1. Edit the spec properties for the KafkaMirrorMaker2 resource.

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

   apiVersion: kafka.strimzi.io/v1alpha1
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker2
   spec:
     version: 2.6.0 1
     replicas: 3 2
     connectCluster: "my-cluster-target" 3
     clusters: 4
     - alias: "my-cluster-source" 5
       authentication: 6
         certificateAndKey:
           certificate: source.crt
           key: source.key
           secretName: my-user-source
         type: tls
       bootstrapServers: my-cluster-source-kafka-bootstrap:9092 7
       tls: 8
         trustedCertificates:
         - certificate: ca.crt
           secretName: my-cluster-source-cluster-ca-cert
     - alias: "my-cluster-target" 9
       authentication: 10
         certificateAndKey:
           certificate: target.crt
           key: target.key
           secretName: my-user-target
         type: tls
       bootstrapServers: my-cluster-target-kafka-bootstrap:9092 11
       config: 12
         config.storage.replication.factor: 1
         offset.storage.replication.factor: 1
         status.storage.replication.factor: 1
         ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 13
         ssl.enabled.protocols: "TLSv1.2"
         ssl.protocol: "TLSv1.2"
         ssl.endpoint.identification.algorithm: HTTPS 14
       tls: 15
         trustedCertificates:
         - certificate: ca.crt
           secretName: my-cluster-target-cluster-ca-cert
     mirrors: 16
     - sourceCluster: "my-cluster-source" 17
       targetCluster: "my-cluster-target" 18
       sourceConnector: 19
         config:
           replication.factor: 1 20
           offset-syncs.topic.replication.factor: 1 21
           sync.topic.acls.enabled: "false" 22
           replication.policy.separator: "" 23
           replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy" 24
       heartbeatConnector: 25
         config:
           heartbeats.topic.replication.factor: 1 26
       checkpointConnector: 27
         config:
           checkpoints.topic.replication.factor: 1 28
       topicsPattern: ".*" 29
       groupsPattern: "group1|group2|group3" 30
     resources: 31
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 32
       type: inline
       loggers:
         connect.root.logger.level: "INFO"
     readinessProbe: 33
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     jvmOptions: 34
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 35
     template: 36
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 37
         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 38
     externalConfiguration: 39
       env:
         - name: AWS_ACCESS_KEY_ID
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsAccessKey
         - name: AWS_SECRET_ACCESS_KEY
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsSecretAccessKey

   1

   The Kafka Connect version.

   2

   The number of replica nodes.

   3

   Cluster alias for Kafka Connect.

   4

   Specification for the Kafka clusters being synchronized.

   5

   Cluster alias for the source Kafka cluster.

   6

   Authentication for the source cluster, using the TLS mechanism, as shown here, using OAuth bearer tokens, or a SASL-based SCRAM-SHA-512 or PLAIN mechanism.

   7

   Bootstrap server for connection to the source Kafka cluster.

   8

   TLS encryption with key names under which TLS certificates are stored in X.509 format for the source Kafka cluster. If certificates are stored in the same secret, it can be listed multiple times.

   9

   Cluster alias for the target Kafka cluster.

   10

   Authentication for the target Kafka cluster is configured in the same way as for the source Kafka cluster.

   11

   Bootstrap server for connection to the target Kafka cluster.

   12

   Kafka Connect configuration. Standard Apache Kafka configuration may be provided, restricted to those properties not managed directly by AMQ Streams.

   13

   SSL properties for external listeners to run with a specific cipher suite for a TLS version.

   14

   Hostname verification is enabled by setting to HTTPS. An empty string disables the verification.

   15

   TLS encryption for the target Kafka cluster is configured in the same way as for the source Kafka cluster.

   16

   MirrorMaker 2.0 connectors.

   17

   Cluster alias for the source cluster used by the MirrorMaker 2.0 connectors.

   18

   Cluster alias for the target cluster used by the MirrorMaker 2.0 connectors.

   19

   Configuration for the MirrorSourceConnector that creates remote topics. The config overrides the default configuration options.

   20

   Replication factor for mirrored topics created at the target cluster.

   21

   Replication factor for the MirrorSourceConnector offset-syncs internal topic that maps the offsets of the source and target clusters.

   22

   When ACL rules synchronization is enabled, ACLs are applied to synchronized topics. The default is true.

   23

   Defines the separator used for the renaming of remote topics.

   24

   Adds a policy that overrides the automatic renaming of remote topics. Instead of prepending the name with the name of the source cluster, the topic retains its original name. This optional setting is useful for active/passive backups and data migration.

   25

   Configuration for the MirrorHeartbeatConnector that performs connectivity checks. The config overrides the default configuration options.

   26

   Replication factor for the heartbeat topic created at the target cluster.

   27

   Configuration for the MirrorCheckpointConnector that tracks offsets. The config overrides the default configuration options.

   28

   Replication factor for the checkpoints topic created at the target cluster.

   29

   Topic replication from the source cluster defined as regular expression patterns. Here we request all topics.

   30

   Consumer group replication from the source cluster defined as regular expression patterns. Here we request three consumer groups by name. You can use comma-separated lists.

   31

   Requests for reservation of supported resources, currently cpu and memory, and limits to specify the maximum resources that can be consumed.

   32

   Specified Kafka Connect loggers and log levels added directly (inline) or indirectly (external) through a ConfigMap. A custom ConfigMap must be placed under the log4j.properties or log4j2.properties key. For the Kafka Connect log4j.rootLogger logger, you can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.

   33

   Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).

   34

   JVM configuration options to optimize performance for the Virtual Machine (VM) running Kafka MirrorMaker.

   35

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

   36

   Template customization. Here a pod is scheduled with anti-affinity, so the pod is not scheduled on nodes with the same hostname.

   37

   Environment variables are also set for distributed tracing using Jaeger.

   38

   Distributed tracing is enabled for Jaeger.

   39

   External configuration for an OpenShift Secret mounted to Kafka MirrorMaker as an environment variable.

2. Create or update the resource:

   oc apply -f <your-file>

Procedure

1. Edit the spec properties for the KafkaBridge resource.

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

   apiVersion: kafka.strimzi.io/v1alpha1
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     replicas: 3 1
     bootstrapServers: my-cluster-kafka-bootstrap:9092 2
     tls: 3
       trustedCertificates:
         - secretName: my-cluster-cluster-cert
           certificate: ca.crt
         - secretName: my-cluster-cluster-cert
           certificate: ca2.crt
     authentication: 4
       type: tls
       certificateAndKey:
         secretName: my-secret
         certificate: public.crt
         key: private.key
     http: 5
       port: 8080
       cors: 6
         allowedOrigins: "https://strimzi.io"
         allowedMethods: "GET,POST,PUT,DELETE,OPTIONS,PATCH"
     consumer: 7
       config:
         auto.offset.reset: earliest
     producer: 8
       config:
         delivery.timeout.ms: 300000
     resources: 9
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 10
       type: inline
       loggers:
         logger.bridge.level: "INFO"
         # enabling DEBUG just for send operation
         logger.send.name: "http.openapi.operation.send"
         logger.send.level: "DEBUG"
     jvmOptions: 11
       "-Xmx": "1g"
       "-Xms": "1g"
     readinessProbe: 12
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     image: my-org/my-image:latest 13
     template: 14
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       bridgeContainer: 15
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"

   1

   The number of replica nodes.

   2

   Bootstrap server for connection to the target Kafka cluster.

   3

   TLS encryption with key names under which TLS certificates are stored in X.509 format for the source Kafka cluster. If certificates are stored in the same secret, it can be listed multiple times.

   4

   Authentication for the Kafka Bridge cluster, using the TLS mechanism, as shown here, using OAuth bearer tokens, or a SASL-based SCRAM-SHA-512 or PLAIN mechanism. By default, the Kafka Bridge connects to Kafka brokers without authentication.

   5

   HTTP access to Kafka brokers.

   6

   CORS access specifying selected resources and access methods. Additional HTTP headers in requests describe the origins that are permitted access to the Kafka cluster.

   7

   Consumer configuration options.

   8

   Producer configuration options.

   9

   Requests for reservation of supported resources, currently cpu and memory, and limits to specify the maximum resources that can be consumed.

   10

   Specified Kafka Bridge loggers and log levels added directly (inline) or indirectly (external) through a ConfigMap. A custom ConfigMap must be placed under the log4j.properties or log4j2.properties key. For the Kafka Bridge loggers, you can set the log level to INFO, ERROR, WARN, TRACE, DEBUG, FATAL or OFF.

   11

   JVM configuration options to optimize performance for the Virtual Machine (VM) running the Kafka Bridge.

   12

   Healthchecks to know when to restart a container (liveness) and when a container can accept traffic (readiness).

   13

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

   14

   Template customization. Here a pod is scheduled with anti-affinity, so the pod is not scheduled on nodes with the same hostname.

   15

   Environment variables are also set for distributed tracing using Jaeger.

2. Create or update the resource:

   oc apply -f KAFKA-BRIDGE-CONFIG-FILE

Procedure

1. Create the ConfigMap.

   You can create the ConfigMap as a YAML file or from a properties file using oc at the command line.

   ConfigMap example with a root logger definition for Kafka:

   kind: ConfigMap
   apiVersion: kafka.strimzi.io/v1beta1
   metadata:
     name: logging-configmap
   data:
     log4j.properties:
       kafka.root.logger.level="INFO"

   From the command line, using a properties file:

   oc create configmap logging-configmap --from-file=log4j.properties

   The properties file defines the logging configuration:

   # Define the logger
   kafka.root.logger.level="INFO"
   # ...

2. Define external logging in the spec of the resource, setting the logging.name to the name of the ConfigMap.

   spec:
     # ...
     logging:
       type: external
       name: logging-configmap

3. Create or update the resource.

   oc apply -f kafka.yaml

1. Mutual TLS authentication (only on the listeners with TLS enabled encryption)
2. SCRAM-SHA-512 authentication
3. OAuth 2.0 token based authentication

Procedure

1. Configure the Kafka resource.

   1. Configure the authorization property for authorization.

   2. Configure the listeners property to create a listener with authentication.

      For example:

      apiVersion: kafka.strimzi.io/v1beta1
      kind: Kafka
      spec:
        kafka:
          # ...
          authorization: 1
            type: simple
            superUsers: 2
              - CN=client_1
              - user_2
              - CN=client_3
          listeners:
            - name: tls
              port: 9093
              type: internal
              tls: true
              authentication:
                type: tls 3
          # ...
        zookeeper:
          # ...

      1

      Authorization enables simple authorization on the Kafka broker using the AclAuthorizer Kafka plugin.

      2

      List of user principals with unlimited access to Kafka. CN is the common name from the client certificate when TLS authentication is used.

      3

      Listener authentication mechanisms may be configured for each listener, and specified as mutual TLS, SCRAM-SHA-512 or token-based OAuth 2.0.

      If you are configuring an external listener, the configuration is dependent on the chosen connection mechanism.

2. Create or update the Kafka resource.

   oc apply -f KAFKA-CONFIG-FILE

   The Kafka cluster is configured with a Kafka broker listener using TLS authentication.

   A service is created for each Kafka broker pod.

   A service is created to serve as the bootstrap address for connection to the Kafka cluster.

   The cluster CA certificate to verify the identity of the kafka brokers is also created with the same name as the Kafka resource.

Procedure

1. Configure the KafkaUser resource.

   For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaUser
   metadata:
     name: my-user
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     authentication: 1
       type: tls
     authorization:
       type: simple 2
       acls:
         - resource:
             type: topic
             name: my-topic
             patternType: literal
           operation: Read
         - resource:
             type: topic
             name: my-topic
             patternType: literal
           operation: Describe
         - resource:
             type: group
             name: my-group
             patternType: literal
           operation: Read

   1

   User authentication mechanism, defined as mutual tls or scram-sha-512.

   2

   Simple authorization, which requires an accompanying list of ACL rules.

2. Create or update the KafkaUser resource.

   oc apply -f USER-CONFIG-FILE

   The user is created, as well as a Secret with the same name as the KafkaUser resource. The Secret contains a private and public key for TLS client authentication.

Procedure

1. Open the Kafka resource.

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

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

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
           networkPolicyPeers:
             - podSelector:
                 matchLabels:
                   app: kafka-client
       # ...
     zookeeper:
       # ...

3. Create or update the resource.

   Use oc apply:

   oc apply -f your-file

1. Deploy an authorization server and configure the deployment to integrate with AMQ Streams
2. Deploy or update the Kafka cluster with Kafka broker listeners configured to use OAuth 2.0
3. Update your Java-based Kafka clients to use OAuth 2.0
4. Update Kafka component clients to use OAuth 2.0

Procedure

1. Access the Red Hat Single Sign-On Admin Console or use the Red Hat Single Sign-On Admin CLI to enable Authorization Services for the Kafka broker client you created when setting up OAuth 2.0 authentication.
2. Use Authorization Services to define resources, authorization scopes, policies, and permissions for the client.
3. Bind the permissions to users and clients by assigning them roles and groups.

4. Configure the Kafka brokers to use Red Hat Single Sign-On authorization by updating the Kafka broker configuration (Kafka.spec.kafka) of your Kafka resource in an editor.

   oc edit kafka my-cluster

5. Configure the Kafka broker kafka configuration to use keycloak authorization, and to be able to access the authorization server and Authorization Services.

   For example:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka
     # ...
     authorization:
       type: keycloak 1
       tokenEndpointUri: <https://<auth-server-address>/auth/realms/external/protocol/openid-connect/token> 2
       clientId: kafka 3
       delegateToKafkaAcls: false 4
       disableTlsHostnameVerification: false 5
       superUsers: 6
         - CN=fred
         - sam
         - CN=edward
       tlsTrustedCertificates: 7
       - secretName: oauth-server-cert
         certificate: ca.crt
       grantsRefreshPeriodSeconds: 60 8
       grantsRefreshPoolSize: 5 9
     #...

   1

   Type keycloak enables Red Hat Single Sign-On authorization.

   2

   URI of the Red Hat Single Sign-On token endpoint. For production, always use HTTPs.

   3

   The client ID of the OAuth 2.0 client definition in Red Hat Single Sign-On that has Authorization Services enabled. Typically, kafka is used as the ID.

   4

   (Optional) Delegate authorization to Kafka AclAuthorizer if access is denied by Red Hat Single Sign-On Authorization Services policies. Default is false.

   5

   (Optional) Disable TLS hostname verification. Default is false.

   6

   (Optional) Designated super users.

   7

   (Optional) Trusted certificates for TLS connection to the authorization server.

   8

   (Optional) The time between two consecutive grants refresh runs. That is the maximum time for active sessions to detect any permissions changes for the user on Red Hat Single Sign-On. The default value is 60.

   9

   (Optional) The number of threads to use to refresh (in parallel) the grants for the active sessions. The default value is 5.

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

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

   oc logs -f ${POD_NAME} -c kafka
   oc get po -w

   The rolling update configures the brokers to use OAuth 2.0 authorization.

8. Verify the configured permissions by accessing Kafka brokers as clients or users with specific roles, making sure they have the necessary access, or do not have the access they are not supposed to have.