步骤

1. 查找控制您要手动更新的 Kafka pod 的 StatefulSet 的名称。

   例如，如果您的 Kafka 集群名为 my-cluster，则对应的 StatefulSet 名为 my-cluster-kafka。

2. 给 OpenShift 中的 StatefulSet 资源标注。例如，使用 oc annotate:

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

3. 等待下一次协调发生（默认为每隔两分钟）。只要协调过程检测到注解的 StatefulSet，则会触发注解的滚动更新。当所有 pod 的滚动更新完成后，注解会从 StatefulSet 中删除。

步骤

1. 查找控制您要手动更新的 ZooKeeper pod 的 StatefulSet 的名称。

   例如，如果您的 Kafka 集群名为 my-cluster，则对应的 StatefulSet 被 命名为 my-cluster-zookeeper。

2. 给 OpenShift 中的 StatefulSet 资源标注。例如，使用 oc annotate:

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

3. 等待下一次协调发生（默认为每隔两分钟）。只要协调过程检测到注解的 StatefulSet，则会触发注解的滚动更新。当所有 pod 的滚动更新完成后，注解会从 StatefulSet 中删除。

步骤

1. 查找您要删除的 Pod 的名称。

   例如，如果集群命名为 cluster-name，pod 被命名为 cluster-name-kafka-index，其中 索引 以零开始，结尾是副本总数。

2. 给 OpenShift 中的 容器集资源标注：

   使用 oc annotate:

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

3. 等待下一次协调，当带有底层持久性卷声明的注解的 pod 被删除后再重新创建。

步骤

1. 查找您要删除的 Pod 的名称。

   例如，如果集群名为 cluster-name，pod 被命名为 cluster-name-zookeeper-index，其中 索引 从零开始，结尾是副本总数。

2. 给 OpenShift 中的 容器集资源标注：

   使用 oc annotate:

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

3. 等待下一次协调，当带有底层持久性卷声明的注解的 pod 被删除后再重新创建。

步骤

1. 将 strimzi.io/force-renew 注解应用到包含您要续订的 CA 证书的 Secret。

   表 2.1. 强制续订证书的 Secret 注解

   证书	Secret	annotate 命令
   集群 CA	kAFKA-CLUSTER-NAME-cluster-ca-cert	oc annotate secret KAFKA-CLUSTER-NAME-cluster-ca-cert strimzi.io/force-renew=true
   客户端 CA	kAFKA-CLUSTER-NAME-clients-ca-cert	oc annotate secret KAFKA-CLUSTER-NAME-clients-ca-cert strimzi.io/force-renew=true

   在下一次协调中，Cluster Operator 会为您注解的 Secret 生成一个新的 CA 证书。如果配置了维护时间窗，Cluster Operator 将在下一个维护时间窗内第一次协调时生成新的 CA 证书。

   客户端应用程序必须重新载入由 Cluster Operator 更新的集群和客户端 CA 证书。

2. 检查 CA 证书有效的周期：

   例如，使用 openssl 命令：

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

   ca-CERTIFICATE-SECRET 是 Secret 的名称，这是用于集群 CA 证书的 KAFKA-CLUSTER-NAME-cluster-ca-cert 和用于客户端 CA 证书的 KAFKA-CLUSTER-NAME-clients-ca-cert。

   ca-CERTIFICATE 是 CA 证书的名称，如 jsonpath={.data.ca\.crt}。

   该命令会返回 notBefore 和 notAfter 日期，这是 CA 证书的有效周期。

   例如，对于集群 CA 证书：

   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. 从 Secret 删除旧证书。

   当组件使用新证书时，旧证书可能仍处于活跃状态。删除旧证书以消除任何潜在的安全风险。

步骤

1. 编辑 KafkaConnect 或 KafkaConnect S2I 资源的 spec 属性。

   您可以配置的属性显示在此示例配置中：

   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

   根据需要，使用 KafkaConnect 或 KafkaConnectS2I。

   2

   为 Kafka Connect 集群启用 KafkaConnectors。

   3

   副本节点的数量。

   4

   Kafka Connect 集群的身份验证，使用 TLS 机制 （如下所示）、使用 OAuth bearer 令牌 或基于 SASL 的 SCRAM-SHA-512 或 PLAIN 机制。默认情况下，Kafka Connect 使用纯文本连接连接到 Kafka 代理。

   5

   用于连接到 Kafka Connect 集群的 bootstrap 服务器。

   6

   使用密钥名称进行 TLS 加密，在其下，TLS 证书以 X.509 格式存储到集群的 X.509 格式。如果证书存储在同一个 secret 中，则可以多次列出证书。

   7

   Kafka Connect worker （ 而不是连接器）配置。标准 Apache Kafka 配置可能会提供，仅限于不是由 AMQ Streams 直接管理的属性。

   8

   使用环境变量（如下所示）或卷 为 Kafka 连接器的外部配置。

   9

   请求保留 支持的资源、当前 cpu 和 memory ，以及限制，以指定可消耗的最大资源。

   10

   指定 Kafka 连接日志记录器和日志级别 直接（内联）或通过ConfigMap 间接（外部）添加。自定义 ConfigMap 必须放在 log4j.properties 或 log4j2.properties 键下。对于 Kafka Connect log4j.rootLogger 日志记录器，您可以将日志级别设置为 INFO、ERROR、WARN、TRACE、DEBUG、FATAL 或 OFF。

   11

   健康检查以了解 何时重新启动容器（存活度）以及容器何时可以接受流量（就绪度）。

   12

   Prometheus metrics，本例中通过 Prometheus JMX 导出器配置启用的 Prometheus 指标。您可以使用 metrics: {} 来在没有进一步配置的情况下启用指标。

   13

   运行 Kafka Connect 的虚拟机(VM)性能优化 JVM 配置选项。

   14

   高级选项：容器镜像配置，建议仅在特殊情况下使用。

   15

   模板自定义.在这里，pod 被调度为反关联性，因此 pod 不会调度到具有相同主机名的节点。

   16

   还 使用 Jaeger 为分布式追踪设置 环境变量。

2. 创建或更新资源：

   oc apply -f KAFKA-CONNECT-CONFIG-FILE

3. 如果为 Kafka Connect 启用了授权，请配置 Kafka Connect 用户以启用对 Kafka Connect consumer 组和主题的访问。

步骤

1. 编辑 KafkaUser 资源中 的授权 属性，以便为用户提供访问权限。

   在以下示例中，使用 字面 名称值为 Kafka Connect 主题和消费者组配置访问权限：

   属性	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. 创建或更新资源。

   oc apply -f KAFKA-USER-CONFIG-FILE

步骤

1. 编辑 KafkaMirrorMaker 资源的 spec 属性。

   您可以配置的属性显示在此示例配置中：

   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

   副本节点的数量。

   2

   为 使用者和生产者引导服务器.

   3

   消费者的组 ID。

   4

   消费者流的数量。

   5

   以毫秒为单位的偏移自动提交间隔。

   6

   使用密钥名称进行 TLS 加密，其中 TLS 证书存储在 X.509 格式供使用者或生产者使用。如果证书存储在同一个 secret 中，则可以多次列出证书。

   7

   对使用者或生产者进行身份验证，使用此处所示的 TLS 机制，使用 OAuth bearer 令牌 或基于 SASL 的 SCRAM-SHA-512 或 PLAIN 机制。

   8

   用于 使用者和 生产者 的 Kafka 配置选项。

   9

   使用特定 密码套件 为 TLS 版本运行外部侦听器的 SSL 属性。

   10

   通过将 设置为 HTTPS 来 启用主机名验证。空字符串将禁用验证。

   11

   如果 abort OnSendFailure 属性 设为 true，则 Kafka MirrorMaker 将退出，容器将在发送失败消息后重启。

   12

   使用特定 密码套件 为 TLS 版本运行外部侦听器的 SSL 属性。

   13

   通过将 设置为 HTTPS 来 启用主机名验证。空字符串将禁用验证。

   14

   从源镜像到目标 Kafka 集群 的主题白名单 。

   15

   请求保留 支持的资源、当前 cpu 和 memory ，以及限制，以指定可消耗的最大资源。

   16

   指定日志记录器和日志级别 通过 ConfigMap 直接（内联）或间接（外部）添加。自定义 ConfigMap 必须放在 log4j.properties 或 log4j2.properties 键下。MirrorMaker 有一个名为 mirrormaker.root.logger 的日志记录器。您可以将日志级别设置为 INFO、ERROR、WARN、TRACE、DEBUG、FATAL 或 OFF。

   17

   健康检查以了解 何时重新启动容器（存活度）以及容器何时可以接受流量（就绪度）。

   18

   Prometheus metrics，本例中通过 Prometheus JMX 导出器配置启用的 Prometheus 指标。您可以使用 metrics: {} 来在没有进一步配置的情况下启用指标。

   19

   用于优化运行 Kafka MirrorMaker 的虚拟机(VM)性能的 JVM 配置选项。

   20

   高级选项：容器镜像配置，建议仅在特殊情况下使用。

   21

   模板自定义.在这里，pod 被调度为反关联性，因此 pod 不会调度到具有相同主机名的节点。

   22

   还 使用 Jaeger 为分布式追踪设置 环境变量。

   23

   为 Jaeger 启用分布式追踪。

   警告

   将 abort OnSendFailure 属性设置为 false 时，制作者会尝试发送主题中的下一个消息。原始消息可能会丢失，因为没有尝试重新发送失败的消息。

2. 创建或更新资源：

   oc apply -f <your-file>

步骤

1. 编辑 KafkaMirrorMaker2 资源的 spec 属性。

   您可以配置的属性显示在此示例配置中：

   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

   Kafka Connect 版本。

   2

   副本节点的数量。

   3

   Kafka Connect 的集群 别名.

   4

   正在同步 的 Kafka 集群的规格。

   5

   源 Kafka 集群的集群 别名。

   6

   使用此处所示的 TLS 机制 （使用 OAuth bearer 令牌 或基于 SASL 的 SCRAM-SHA-512 或 PLAIN 机制）对源集群进行身份验证。

   7

   用于 连接到源 Kafka 集群的 bootstrap 服务器。

   8

   使用源 Kafka 集群的 TLS 证书以 X.509 格式存储的密钥名称进行 TLS 加密。如果证书存储在同一个 secret 中，则可以多次列出证书。

   9

   目标 Kafka 集群的集群 别名。

   10

   目标 Kafka 集群的身份验证的配置方式与源 Kafka 集群相同。

   11

   用于 连接到目标 Kafka 集群的 bootstrap 服务器。

   12

   Kafka Connect 配置.标准 Apache Kafka 配置可能会提供，仅限于不是由 AMQ Streams 直接管理的属性。

   13

   使用特定 密码套件 为 TLS 版本运行外部侦听器的 SSL 属性。

   14

   通过将 设置为 HTTPS 来 启用主机名验证。空字符串将禁用验证。

   15

   目标 Kafka 集群的 TLS 加密配置方式与源 Kafka 集群相同。

   16

   MirrorMaker 2.0 连接器.

   17

   MirrorMaker 2.0 连接器使用的源集群的集群 别名。

   18

   MirrorMaker 2.0 连接器使用的目标集群的集群 别名。

   19

   创建远程主题的 MirrorSourceConnector配置。配置 会覆盖默认配置选项。

   20

   在目标集群中创建的已镜像主题的复制因素。

   21

   MirrorSourceConnector offset-syncs 内部主题的复制因素，用于映射源和目标集群的偏移。

   22

   启用 ACL 规则同步 时，将应用 ACL 来同步主题。默认值是 true。

   23

   定义用于重命名远程主题的分隔符。

   24

   添加可覆盖远程主题自动重命名的策略。该主题不会用源集群的名称来附加名称，而是保留其原始名称。此可选设置对主动/被动备份和数据迁移很有用。

   25

   执行连接检查 的 MirrorHeartbeatConnector配置。配置 会覆盖默认配置选项。

   26

   在目标集群中创建的心跳主题的复制因素。

   27

   跟踪偏移的 MirrorCheckpointConnector配置。配置 会覆盖默认配置选项。

   28

   在目标集群中创建的检查点主题的复制因素。

   29

   从 定义为正则表达式模式的 源集群进行主题复制。这里我们请求所有主题。

   30

   来自 定义为正则表达式模式的 源集群的使用者组复制。在这里，我们按名称请求三个消费者组。您可以使用逗号分隔的列表。

   31

   请求保留 支持的资源、当前 cpu 和 memory ，以及限制，以指定可消耗的最大资源。

   32

   指定 Kafka 连接日志记录器和日志级别 直接（内联）或通过ConfigMap 间接（外部）添加。自定义 ConfigMap 必须放在 log4j.properties 或 log4j2.properties 键下。对于 Kafka Connect log4j.rootLogger 日志记录器，您可以将日志级别设置为 INFO、ERROR、WARN、TRACE、DEBUG、FATAL 或 OFF。

   33

   健康检查以了解 何时重新启动容器（存活度）以及容器何时可以接受流量（就绪度）。

   34

   用于优化运行 Kafka MirrorMaker 的虚拟机(VM)性能的 JVM 配置选项。

   35

   高级选项：容器镜像配置，建议仅在特殊情况下使用。

   36

   模板自定义.在这里，pod 被调度为反关联性，因此 pod 不会调度到具有相同主机名的节点。

   37

   还 使用 Jaeger 为分布式追踪设置 环境变量。

   38

   为 Jaeger 启用分布式追踪。

   39

   作为环境变量挂载到 Kafka MirrorMaker 的 OpenShift Secret 的外部配置。

2. 创建或更新资源：

   oc apply -f <your-file>

步骤

1. 编辑 KafkaBridge 资源的 spec 属性。

   您可以配置的属性显示在此示例配置中：

   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

   副本节点的数量。

   2

   用于 连接到目标 Kafka 集群的 bootstrap 服务器。

   3

   使用源 Kafka 集群的 TLS 证书以 X.509 格式存储的密钥名称进行 TLS 加密。如果证书存储在同一个 secret 中，则可以多次列出证书。

   4

   Kafka Bridge 集群的身份验证，使用 TLS 机制 （此处所示）、使用 OAuth bearer 令牌 或基于 SASL 的 SCRAM-SHA-512 或 PLAIN 机制。默认情况下，Kafka Bridge 在不进行身份验证的情况下连接到 Kafka 代理。

   5

   对 Kafka 代理的 HTTP 访问.

   6

   CORS 访问 指定选定的资源和访问方法.请求中的附加 HTTP 标头描述了允许访问 Kafka 集群的原始数据。

   7

   使用者配置选项.

   8

   制作者配置选项.

   9

   请求保留 支持的资源、当前 cpu 和 memory ，以及限制，以指定可消耗的最大资源。

   10

   指定 Kafka Bridge 日志记录器和日志级别 直接（内联）或通过ConfigMap 间接（外部）添加。自定义 ConfigMap 必须放在 log4j.properties 或 log4j2.properties 键下。对于 Kafka Bridge loggers，您可以将日志级别设置为 INFO、ERROR、WARN、TRACE、DEBUG、FATAL 或 OFF。

   11

   运行 Kafka 网桥的虚拟机(VM)的 JVM 配置选项 优化性能。

   12

   健康检查以了解 何时重新启动容器（存活度）以及容器何时可以接受流量（就绪度）。

   13

   高级选项：容器镜像配置，建议仅在特殊情况下使用。

   14

   模板自定义.在这里，pod 被调度为反关联性，因此 pod 不会调度到具有相同主机名的节点。

   15

   还 使用 Jaeger 为分布式追踪设置 环境变量。

2. 创建或更新资源：

   oc apply -f KAFKA-BRIDGE-CONFIG-FILE

步骤

1. 创建 ConfigMap。

   您可以使用位于命令行的 oc 以 YAML 文件或属性文件创建 ConfigMap。

   带有 Kafka 的根日志记录器定义的 ConfigMap 示例：

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

   在命令行中使用属性文件：

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

   属性文件定义日志配置：

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

2. 在资源的 spec 中定义 外部 日志记录，将 logging.name 设置为 ConfigMap 的名称。

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

3. 创建或更新资源。

   oc apply -f kafka.yaml

1. 双向 TLS 身份验证（仅在启用了 TLS 加密的监听器中）
2. SCRAM-SHA-512 验证
3. 基于 OAuth 2.0 令牌的身份验证

步骤

1. 配置 Kafka 资源。

   1. 配置 授权 属性。

   2. 配置 listens 属性，以创建具有身份验证的侦听器。

      例如：

      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

      授权启用使用 AclAuthorizer Kafka 插件的 Kafka 代理 的简单 授权。

      2

      对 Kafka 拥有无限访问权限的用户主体列表。在使用 TLS 身份验证时，CN 是客户端证书的常用名称。

      3

      监听器身份验证机制可以为每个侦听器配置，并 指定为 mutual TLS、SCRAM-SHA-512 或基于令牌的 OAuth 2.0。

      如果您要配置外部监听程序，其配置取决于所选的连接机制。

2. 创建或更新 Kafka 资源。

   oc apply -f KAFKA-CONFIG-FILE

   使用 TLS 身份验证配置 Kafka 代理监听程序。

   为每个 Kafka 代理 pod 创建一个服务。

   创建一个服务作为连接到 Kafka 集群的 bootstrap 地址。

   也使用与 Kafka 资源相同的名称来创建用于验证 kafka 代理身份的集群 CA 证书。

步骤

1. 配置 KafkaUser 资源。

   例如：

   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

   用户身份验证机制，定义为 mutual tls 或 scr am-sha-512。

   2

   简单授权，这需要附带的 ACL 规则列表。

2. 创建或更新 KafkaUser 资源。

   oc apply -f USER-CONFIG-FILE

   已创建用户，以及名称与 KafkaUser 资源相同的 Secret。Secret 包含用于 TLS 客户端身份验证的私钥和公钥。

步骤

1. 打开 Kafka 资源。

2. 在 networkPolicyPeers 属性中，定义允许访问 Kafka 集群的应用程序 pod 或命名空间。

   例如，将 a tls 侦听程序配置为只允许来自标签 app 设置为 kafka-client 的应用程序 pod 的连接：

   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. 创建或更新资源。

   使用 oc apply:

   oc apply -f your-file

1. 部署授权服务器并配置部署以与 AMQ 流集成
2. 使用配置为使用 OAuth 2.0 的 Kafka 代理监听程序部署或更新 Kafka 集群
3. 更新基于 Java 的 Kafka 客户端以使用 OAuth 2.0
4. 更新 Kafka 组件客户端以使用 OAuth 2.0

步骤

1. 访问 Red Hat Single Sign-On Admin 控制台，或使用 Red Hat Single Sign-On Admin CLI 为您在设置 OAuth 2.0 身份验证时创建的 Kafka 代理客户端启用授权服务。
2. 使用授权服务为客户端定义资源、授权范围、策略和权限。
3. 通过为用户和组分配角色和组，将权限绑定至用户和组。

4. 通过在 编辑器 中更新 Kafka 代理配置(Kafka.spec.kafka)，将 Kafka代理配置为使用 Red Hat Single Sign-On 授权。

   oc edit kafka my-cluster

5. 将 Kafka 代理 kafka 配置配置为使用 keycloak 授权，并可以访问授权服务器和授权服务。

   例如：

   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

   类型 keycloak 启用红帽单点登录授权。

   2

   红帽单点登录令牌端点的 URI。对于生产环境，始终使用 HTTPs。

   3

   启用了授权服务的 Red Hat Single Sign-On 中的 OAuth 2.0 客户端定义的客户端 ID。通常, kafka 用作 ID。

   4

   （可选）如果 Red Hat Single Sign-On Authorization Services 策略拒绝访问，则会向 Kafka AclAuthorizer 委派授权。默认为 false。

   5

   （可选）禁用 TLS 主机名验证。默认为 false。

   6

   （可选）指定的 超级用户.

   7

   （可选）用于 TLS 连接到授权服务器的受信任证书。

   8

   （可选）两个连续的时间允许刷新运行。这是活跃会话的最大时间，用于检测 Red Hat Single Sign-On 上用户的任何权限更改。默认值为 60。

   9

   （可选）用于（并行）活动会话授权的线程数量。默认值为 5。

6. 保存并退出编辑器，然后等待滚动更新完成。

7. 检查日志中的更新，或者查看 pod 状态转换：

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

   滚动更新将代理配置为使用 OAuth 2.0 授权。

8. 以具有特定角色的客户端或用户访问 Kafka 代理，确保他们具有必要的访问权限，或者没有他们应该拥有的访问权限，来验证配置的权限。