在 OpenShift 中使用 AMQ Streams

流程

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       replicas: 3 1
       version: 2.8.0 2
       logging: 3
         type: inline
         loggers:
           kafka.root.logger.level: "INFO"
       resources: 4
         requests:
           memory: 64Gi
           cpu: "8"
         limits:
           memory: 64Gi
           cpu: "12"
       readinessProbe: 5
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       jvmOptions: 6
         -Xms: 8192m
         -Xmx: 8192m
       image: my-org/my-image:latest 7
       listeners: 8
         - name: plain 9
           port: 9092 10
           type: internal 11
           tls: false 12
           configuration:
             useServiceDnsDomain: true 13
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication: 14
             type: tls
         - name: external 15
           port: 9094
           type: route
           tls: true
           configuration:
             brokerCertChainAndKey: 16
               secretName: my-secret
               certificate: my-certificate.crt
               key: my-key.key
       authorization: 17
         type: simple
       config: 18
         auto.create.topics.enable: "false"
         offsets.topic.replication.factor: 3
         transaction.state.log.replication.factor: 3
         transaction.state.log.min.isr: 2
         log.message.format.version: 2.8
         inter.broker.protocol.version: 2.8
         ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 19
         ssl.enabled.protocols: "TLSv1.2"
         ssl.protocol: "TLSv1.2"
       storage: 20
         type: persistent-claim 21
         size: 10000Gi 22
       rack: 23
         topologyKey: topology.kubernetes.io/zone
       metricsConfig: 24
         type: jmxPrometheusExporter
         valueFrom:
           configMapKeyRef: 25
             name: my-config-map
             key: my-key
       # ...
     zookeeper: 26
       replicas: 3 27
       logging: 28
         type: inline
         loggers:
           zookeeper.root.logger: "INFO"
       resources:
         requests:
           memory: 8Gi
           cpu: "2"
         limits:
           memory: 8Gi
           cpu: "2"
       jvmOptions:
         -Xms: 4096m
         -Xmx: 4096m
       storage:
         type: persistent-claim
         size: 1000Gi
       metricsConfig:
         # ...
     entityOperator: 29
       tlsSidecar: 30
         resources:
           requests:
             cpu: 200m
             memory: 64Mi
           limits:
             cpu: 500m
             memory: 128Mi
       topicOperator:
         watchedNamespace: my-topic-namespace
         reconciliationIntervalSeconds: 60
         logging: 31
           type: inline
           loggers:
             rootLogger.level: "INFO"
         resources:
           requests:
             memory: 512Mi
             cpu: "1"
           limits:
             memory: 512Mi
             cpu: "1"
       userOperator:
         watchedNamespace: my-topic-namespace
         reconciliationIntervalSeconds: 60
         logging: 32
           type: inline
           loggers:
             rootLogger.level: INFO
         resources:
           requests:
             memory: 512Mi
             cpu: "1"
           limits:
             memory: 512Mi
             cpu: "1"
     kafkaExporter: 33
       # ...
     cruiseControl: 34
       # ...
       tlsSidecar: 35
       # ...

   1

   副本节点的数量。如果您的集群已经定义了主题，您可以 扩展集群。

   2

   Kafka 版本，可按照 升级过程将其改为受支持的 版本。

   3

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

   4

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

   5

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

   6

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

   7

   ADVANCED OPTION： 容器镜像配置，只在特殊情况下推荐这样做。

   8

   侦听器配置客户端如何通过 bootstrap 地址连接到 Kafka 集群。侦听器 配置为 内部或外部 侦听器，以便从 OpenShift 集群内部或外部进行连接。

   9

   用于标识侦听器的名称。在 Kafka 集群中必须是唯一的。

   10

   Kafka 内侦听器使用的端口号。在给定 Kafka 集群中，端口号必须是唯一的。允许的端口号是 9092 及以上，但端口 9404 和 9999 除外，它们已用于 Prometheus 和 JMX。根据监听程序类型，端口号可能与连接 Kafka 客户端的端口号不同。

   11

   侦听器类型指定为 内部，或用于外部监听器，作为 路由、负载均衡器、 节点端口 或 入口。

   12

   为每个侦听器启用 TLS 加密。默认为 false。路由 监听程序不需要 TLS 加密。

   13

   定义是否分配了包括集群服务后缀（通常. cluster.local）在内的完全限定 DNS 名称。

   14

   侦听器身份验证机制 指定为 mutual TLS、SCRAM-SHA-512 或基于令牌的 OAuth 2.0。

   15

   外部监听程序配置指定 Kafka 集群如何公开 OpenShift 外部，如通过 路由、loadbalancer 或 nodeport。

   16

   由外部 证书颁发机构管理的 Kafka 侦听器 证书的可选配置。brokerCertChainAndKey 指定包含服务器证书和私钥的 Secret。您可以在任何启用 TLS 加密的监听程序上配置 Kafka 侦听程序证书。

   17

   授权在 Kafka 代理上启用简单、OAUTH 2.0 或 OPA 授权。简单授权使用 AclAuthorizer Kafka 插件。

   18

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

   19

   启用 TLS 加密的监听器的 SSL 属性，以启用特定的 密码套件 或 TLS 版本。

   20

   存储 被配置为 临时、持久声明或 jbod。

   21

   持久性卷的存储大小可能会增加，并 可向 JBOD 存储添加额外的卷。

   22

   持久性存储具有 额外的配置选项，如用于动态卷置备的存储 id 和 class。

   23

   将机架感知 配置为在不同机架之间分布副本。topologykey 必须与集群节点标签匹配。

   24

   启用了 Prometheus 指标。在本例中，为 Prometheus JMX Exporter（默认指标导出器）配置了指标数据。

   25

   通过 Prometheus JMX Exporter 将指标导出到 Grafana 仪表板的 Prometheus 规则，这可以通过引用包含 Prometheus JMX 导出器配置的 ConfigMap 来启用。您可以使用对包含 metricsConfig.valueFrom.configMapKeyRef.key 下的空文件的 ConfigMap 的引用来启用指标，而无需进一步配置。

   26

   zookeeper 特定的配置，其中包含与 Kafka 配置类似的属性。

   27

   ZooKeeper 节点的数量。zookeeper 集群或ensemble 通常使用奇数的节点（通常为三个、五个或 7 个）运行。大多数节点必须可用，才能保持有效的仲裁。如果 ZooKeeper 集群丢失其仲裁，它将停止响应客户端，Kafka 代理将停止工作。拥有稳定且高度可用的 ZooKeeper 集群对于 AMQ 流至关重要。

   28

   指定 ZooKeeper 日志记录器和日志级别。

   29

   实体 Operator 配置，用于指定 Topic Operator 和 User Operator 的配置。

   30

   实体操作器 TLS sidecar 配置.实体 Operator 使用 TLS sidecar 与 ZooKeeper 进行安全通信。

   31

   指定 Topic Operator 日志记录器和日志级别。这个示例使用 内联 日志记录。

   32

   指定 的用户 Operator 日志记录器和日志级别。

   33

   Kafka Exporter 配置.Kafka Exporter 是一个可选组件，用于从 Kafka 代理提取指标数据，特别是消费者滞后数据。

   34

   Cruise Control 的可选配置，用于 重新平衡 Kafka 集群。

   35

   插入控制 TLS sidecar 配置.cruise Control 使用 TLS sidecar 与 ZooKeeper 进行安全通信。

2. 创建或更新资源：

   oc apply -f KAFKA-CONFIG-FILE

流程

1. 使用 OpenShift 控制台打开终端，或者从 CLI 运行 exec 命令。

   例如：

   oc exec -ti my-cluster-zookeeper-0 -- bin/kafka-topics.sh --list --zookeeper localhost:12181

   务必使用 localhost:12181。

   现在，您可以向 ZooKeeper 运行 Kafka 命令。

流程

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. 编辑 KafkaConnect 或 KafkaConnect S2I 资源的 spec 属性。

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

   apiVersion: kafka.strimzi.io/v1beta2
   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
     build: 8
       output: 9
         type: docker
         image: my-registry.io/my-org/my-connect-cluster:latest
         pushSecret: my-registry-credentials
       plugins: 10
         - name: debezium-postgres-connector
           artifacts:
             - type: tgz
               url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/1.3.1.Final/debezium-connector-postgres-1.3.1.Final-plugin.tar.gz
               sha512sum: 962a12151bdf9a5a30627eebac739955a4fd95a08d373b86bdcea2b4d0c27dd6e1edd5cb548045e115e33a9e69b1b2a352bee24df035a0447cb820077af00c03
         - name: camel-telegram
           artifacts:
             - type: tgz
               url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.7.0/camel-telegram-kafka-connector-0.7.0-package.tar.gz
               sha512sum: a9b1ac63e3284bea7836d7d24d84208c49cdf5600070e6bd1535de654f6920b74ad950d51733e8020bf4187870699819f54ef5859c7846ee4081507f48873479
     externalConfiguration: 11
       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: 12
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 13
       type: inline
       loggers:
         log4j.rootLogger: "INFO"
     readinessProbe: 14
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     metricsConfig: 15
       type: jmxPrometheusExporter
       valueFrom:
         configMapKeyRef:
           name: my-config-map
           key: my-key
     jvmOptions: 16
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 17
     rack:
       topologyKey: topology.kubernetes.io/zone 18
     template: 19
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 20
         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

   用于自动使用连接器插件构建容器镜像的 构建配置属性。

   9

   （必需）配置推送新镜像的容器注册表。

   10

   （必需）要添加到新容器镜像的连接器插件及其工件列表。每个插件必须至少配置一个 工件。

   11

   使用环境变量（如下所示）或卷 为 Kafka 连接器的外部配置。您还可以使用 Kubernetes 配置提供程序 从外部来源加载配置值。

   12

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

   13

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

   14

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

   15

   Prometheus metrics，通过引用包含本例中 Prometheus JMX 导出器配置的 ConfigMap 来启用它。您可以使用对包含 metricsConfig.valueFrom.configMapKeyRef.key 下的空文件的 ConfigMap 的引用来启用指标，而无需进一步配置。

   16

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

   17

   ADVANCED OPTION： 容器镜像配置，只在特殊情况下推荐这样做。

   18

   将机架感知 配置为在不同机架之间分布副本。topologykey 必须与集群节点标签匹配。

   19

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

   20

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

2. 创建或更新资源：

   oc apply -f KAFKA-CONNECT-CONFIG-FILE

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

流程

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

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

   属性	名称
   offset.storage.topic	connect-cluster-offsets
   status.storage.topic	connect-cluster-status
   config.storage.topic	connect-cluster-configs
   group	connect-cluster

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-user
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     # ...
     authorization:
       type: simple
       acls:
         # access to offset.storage.topic
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           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. 查找控制您要重启的 Kafka 连接器的 KafkaConnector 自定义资源的名称：

   oc get KafkaConnector

2. 要重启连接器，请注解 OpenShift 中的 KafkaConnector 资源。例如，使用 oc annotate:

   oc annotate KafkaConnector KAFKACONNECTOR-NAME strimzi.io/restart=true

3. 等待下一次协调发生（默认为每隔两分钟）。

   Kafka 连接器重启，只要协调过程检测到注解。当 Kafka Connect 接受重启请求时，该注解会从 KafkaConnector 自定义资源中删除。

流程

1. 查找控制您要重启的 Kafka 连接器任务的 KafkaConnector 自定义资源的名称：

   oc get KafkaConnector

2. 从 KafkaConnector 自定义资源中找到要重启的任务 ID。任务 ID 是非负整数，从 0 开始。

   oc describe KafkaConnector KAFKACONNECTOR-NAME

3. 要重启连接器任务，请注解 OpenShift 中的 KafkaConnector 资源。例如，使用 oc annotate 重启任务 0:

   oc annotate KafkaConnector KAFKACONNECTOR-NAME strimzi.io/restart-task=0

4. 等待下一次协调发生（默认为每隔两分钟）。

   Kafka 连接器任务重启，只要协调过程检测到注解。当 Kafka Connect 接受重启请求时，该注解会从 KafkaConnector 自定义资源中删除。

流程

1. 使用与用于 Kafka connectS2I 资源的名称相同的名称创建新的 Kafka Connect 自定义资源。
2. 将 KafkaConnectS2I 资源属性复制到 KafkaConnect 资源。

3. 如果指定，请确保使用相同的 spec.config 属性：

   • group.id
   • offset.storage.topic
   • config.storage.topic

   • status.storage.topic

     如果未指定这些属性，则使用默认值。在这种情况下，也将其从 KafkaConnect 资源配置中退出。

   现在，将特定于 KafkaConnect 资源的配置添加到新资源中。

4. 添加 构建配置，以配置您要添加到 Kafka Connect 部署中的所有连接器和其他库。

   注意

   或者，您也可以使用连接器手动构建新镜像，并使用 .spec.image 属性指定它。

5. 删除旧的 KafkaConnectS2I 资源：

   oc delete -f MY-KAFKA-CONNECT-S2I-CONFIG-FILE

   将 MY-KAFKA-CONNECT-S2I-CONFIG-FILE 替换为包含 KafkaConnectS2I 资源配置的文件名。

   另外，您可以指定资源的名称：

   oc delete kafkaconnects2i MY-KAFKA-CONNECT-S2I

   将 MY-KAFKA-CONNECT-S2I 替换为 KafkaConnectS2I 资源的名称。

   等待 Kafka Connect with S2I 部署被删除，并且 pod 已被删除。

   警告

   不得删除任何其他资源。

6. 部署新的 KafkaConnect 资源：

   oc apply -f MY-KAFKA-CONNECT-CONFIG-FILE

   将 MY-KAFKA-CONNECT-CONFIG-FILE 替换为包含新 KafkaConnect 资源配置的文件名称。

   等待新镜像构建好，部署创建好，并且容器集已启动。

7. 如果您使用 KafkaConnector 资源来管理 Kafka Connect 连接器，请检查所有预期的连接器是否存在并正在运行：

   oc get kctr --selector strimzi.io/cluster=MY-KAFKA-CONNECT-CLUSTER -o name

   将 MY-KAFKA-CONNECT-CLUSTER 替换为 Kafka Connect 集群的名称。

   通过 Kafka Connect 存储自动恢复连接器。即使您使用 Kafka Connect REST API 管理它们，您不应该手动重新创建它们。

流程

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   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
     include: "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
     metricsConfig: 18
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: my-config-map
          key: my-key
     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 导出器配置的 ConfigMap 来启用它。您可以使用对包含 metricsConfig.valueFrom.configMapKeyRef.key 下的空文件的 ConfigMap 的引用来启用指标，而无需进一步配置。

   19

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

   20

   ADVANCED OPTION： 容器镜像配置，只在特殊情况下推荐这样做。

   21

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

   22

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

   23

   为 Jaeger 启用分布式追踪。

   警告

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

2. 创建或更新资源：

   oc apply -f <your-file>

流程

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker2
   spec:
     version: 2.8.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
         tasksMax: 10 20
         config:
           replication.factor: 1 21
           offset-syncs.topic.replication.factor: 1 22
           sync.topic.acls.enabled: "false" 23
           refresh.topics.interval.seconds: 60 24
           replication.policy.separator: "" 25
           replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy" 26
       heartbeatConnector: 27
         config:
           heartbeats.topic.replication.factor: 1 28
       checkpointConnector: 29
         config:
           checkpoints.topic.replication.factor: 1 30
           refresh.groups.interval.seconds: 600 31
           sync.group.offsets.enabled: true 32
           sync.group.offsets.interval.seconds: 60 33
           emit.checkpoints.interval.seconds: 60 34
           replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"
       topicsPattern: ".*" 35
       groupsPattern: "group1|group2|group3" 36
     resources: 37
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 38
       type: inline
       loggers:
         connect.root.logger.level: "INFO"
     readinessProbe: 39
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     jvmOptions: 40
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 41
     template: 42
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 43
         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 44
     externalConfiguration: 45
       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 和 Mirror Maker 2.0 版本，版本 将始终相同。

   2

   副本节点的数量。

   3

   Kafka Connect 的 Kafka 集群别名，它必须指定 目标 Kafka 集群。Kafka 集群由 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

   连接器可以创建的任务数量上限。任务处理数据复制，并并行运行。如果基础架构支持处理开销，增加这个值可以提高吞吐量。Kafka Connect 在集群成员之间分发任务。如果任务数量超过工作程序，则为工作程序分配多项任务。对于接收器连接器，旨在为每个主题分区使用一个任务。对于源连接器，可以并行运行的任务数量也可能取决于外部系统。如果无法实现并行性，连接器会创建少于任务的最大数量。

   21

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

   22

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

   23

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

   24

   可选设置，可更改新主题的检查频率。默认为每 10 分钟检查一次。

   25

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

   26

   添加可覆盖远程主题自动重命名的策略。该主题不会用源集群的名称来附加名称，而是保留其原始名称。此可选设置对主动/被动备份和数据迁移很有用。要配置主题偏移同步，还必须为 checkpointConnector.config 设置此属性。

   27

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

   28

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

   29

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

   30

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

   31

   可选设置，以更改新使用者组的检查频率。默认为每 10 分钟检查一次。

   32

   可选设置来同步消费者组偏移，这对于主动/被动配置中的恢复非常有用。默认情况下不启用同步。

   33

   如果启用了消费者组偏移的同步，您可以调整同步的频率。

   34

   调整检查偏移跟踪的频率。如果更改了偏移同步的频率，您可能还需要调整这些检查的频率。

   35

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

   36

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

   37

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

   38

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

   39

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

   40

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

   41

   ADVANCED OPTION： 容器镜像配置，只在特殊情况下推荐这样做。

   42

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

   43

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

   44

   为 Jaeger 启用分布式追踪。

   45

   作为环境变量挂载到 Kafka MirrorMaker 的 OpenShift Secret 的外部配置。您还可以使用 Kubernetes 配置提供程序 从外部来源加载配置值。

2. 创建或更新资源：

   oc apply -f MIRRORMAKER-CONFIGURATION-FILE

流程

1. 查找控制您要重启的 Kafka MirrorMaker 2.0 连接器的 KafkaMirrorMaker2 自定义资源的名称：

   oc get KafkaMirrorMaker2

2. 从 Kafka MirrorMaker2 自定义资源中找到要重启的 Kafka MirrorMaker 2.0 连接器的名称。

   oc describe KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME

3. 要重启连接器，请注解 OpenShift 中的 KafkaMirrorMaker2 资源。在本例中，oc annotate 会重启名为 my-source->my-target.MirrorSourceConnector 的连接器：

   oc annotate KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME "strimzi.io/restart-connector=my-source->my-target.MirrorSourceConnector"

4. 等待下一次协调发生（默认为每隔两分钟）。

   Kafka MirrorMaker 2.0 连接器重启，只要协调过程检测到注解。当重启请求被接受时，该注解会从 KafkaMirrorMaker2 自定义资源中删除。

流程

1. 查找控制您要重启的 Kafka MirrorMaker 2.0 连接器的 KafkaMirrorMaker2 自定义资源的名称：

   oc get KafkaMirrorMaker2

2. 查找 Kafka MirrorMaker 2.0 连接器的名称，以及从 KafkaMirrorMaker2 自定义资源重启的任务 ID。任务 ID 是非负整数，从 0 开始。

   oc describe KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME

3. 要重启连接器任务，请注解 OpenShift 中的 KafkaMirrorMaker2 资源。在本例中，oc annotate restarts task 0 of a connector of my-source->my-target.MirrorSourceConnector:

   oc annotate KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME "strimzi.io/restart-connector-task=my-source->my-target.MirrorSourceConnector:0"

4. 等待下一次协调发生（默认为每隔两分钟）。

   Kafka MirrorMaker 2.0 连接器任务重启，只要协调过程检测到注解。当重启任务请求被接受时，注解会从 KafkaMirrorMaker2 自定义资源中删除。

流程

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   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

   ADVANCED OPTION： 容器镜像配置，只在特殊情况下推荐这样做。

   14

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

   15

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

2. 创建或更新资源：

   oc apply -f KAFKA-BRIDGE-CONFIG-FILE

流程

1. 编辑指定集群部署的资源中的 affinity 属性。要确保 Kafka 代理或 ZooKeeper 节点没有共享 worker 节点，请使用 strimzi.io/name 标签。将 topologyKey 设置为 kubernetes.io/hostname，以指定所选 pod 没有调度到具有相同主机名的节点。这仍然允许同一 worker 节点由单个 Kafka 代理和单个 ZooKeeper 节点共享。例如：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       template:
         pod:
           affinity:
             podAntiAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 - labelSelector:
                     matchExpressions:
                       - key: strimzi.io/name
                         operator: In
                         values:
                           - CLUSTER-NAME-kafka
                   topologyKey: "kubernetes.io/hostname"
       # ...
     zookeeper:
       # ...
       template:
         pod:
           affinity:
             podAntiAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 - labelSelector:
                     matchExpressions:
                       - key: strimzi.io/name
                         operator: In
                         values:
                           - CLUSTER-NAME-zookeeper
                   topologyKey: "kubernetes.io/hostname"
       # ...

   Where CLUSTER-NAME 是 Kafka 自定义资源的名称。

2. 如果您甚至希望确保 Kafka 代理和 ZooKeeper 节点不共享相同的 worker 节点，请使用 strimzi.io/cluster 标签。例如：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       template:
         pod:
           affinity:
             podAntiAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 - labelSelector:
                     matchExpressions:
                       - key: strimzi.io/cluster
                         operator: In
                         values:
                           - CLUSTER-NAME
                   topologyKey: "kubernetes.io/hostname"
       # ...
     zookeeper:
       # ...
       template:
         pod:
           affinity:
             podAntiAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 - labelSelector:
                     matchExpressions:
                       - key: strimzi.io/cluster
                         operator: In
                         values:
                           - CLUSTER-NAME
                   topologyKey: "kubernetes.io/hostname"
       # ...

   Where CLUSTER-NAME 是 Kafka 自定义资源的名称。

3. 创建或更新资源。

   oc apply -f KAFKA-CONFIG-FILE

流程

1. 编辑指定集群部署的资源中的 affinity 属性。使用标签指定不应调度到同一节点上的 pod。topologyKey 应设置为 kubernetes.io/hostname，以指定不应将所选 pod 调度到具有相同主机名的节点。例如：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       template:
         pod:
           affinity:
             podAntiAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 - labelSelector:
                     matchExpressions:
                       - key: application
                         operator: In
                         values:
                           - postgresql
                           - mongodb
                   topologyKey: "kubernetes.io/hostname"
       # ...
     zookeeper:
       # ...

2. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f KAFKA-CONFIG-FILE

流程

1. 标记调度 AMQ Streams 组件的节点。

   可以使用 oc label 来完成此操作：

   oc label node NAME-OF-NODE node-type=fast-network

   或者，某些现有标签可以被重复使用。

2. 编辑指定集群部署的资源中的 affinity 属性。例如：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       template:
         pod:
           affinity:
             nodeAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 nodeSelectorTerms:
                   - matchExpressions:
                     - key: node-type
                       operator: In
                       values:
                       - fast-network
       # ...
     zookeeper:
       # ...

3. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f KAFKA-CONFIG-FILE

流程

1. 选择应当用作专用的节点。
2. 确保这些节点上未调度工作负载。

3. 在所选节点上设置污点：

   可以使用 oc adm taint 来完成此操作：

   oc adm taint node NAME-OF-NODE dedicated=Kafka:NoSchedule

4. 另外，也向所选节点添加标签。

   可以使用 oc label 来完成此操作：

   oc label node NAME-OF-NODE dedicated=Kafka

5. 编辑 指定集群部署的资源中的关联性 和 容限 属性。

   例如：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       template:
         pod:
           tolerations:
             - key: "dedicated"
               operator: "Equal"
               value: "Kafka"
               effect: "NoSchedule"
           affinity:
             nodeAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 nodeSelectorTerms:
                 - matchExpressions:
                   - key: dedicated
                     operator: In
                     values:
                     - Kafka
       # ...
     zookeeper:
       # ...

6. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f KAFKA-CONFIG-FILE

流程

1. 创建 ConfigMap。

   您可以将 ConfigMap 作为 YAML 文件或属性文件创建。

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

   kind: ConfigMap
   apiVersion: v1
   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.valueFrom.configMapKeyRef.name 设置为 ConfigMap 的名称，valueFrom.configMapKeyRef.key 设置为此 ConfigMap 中的键。

   spec:
     # ...
     logging:
       type: external
       valueFrom:
         configMapKeyRef:
           name: logging-configmap
           key: log4j.properties

3. 创建或更新资源。

   oc apply -f KAFKA-CONFIG-FILE

流程

1. 更新 050-ConfigMap-strimzi-cluster-operator.yaml 文件，将过滤器属性添加到 ConfigMap 中。

   在本例中，过滤器属性仅为 my-kafka-cluster Kafka 集群返回日志：

   kind: ConfigMap
   apiVersion: v1
   metadata:
     name: strimzi-cluster-operator
   data:
     log4j2.properties:
       #...
       appender.console.filter.filter1.type=MarkerFilter
       appender.console.filter.filter1.onMatch=ACCEPT
       appender.console.filter.filter1.onMismatch=DENY
       appender.console.filter.filter1.marker=Kafka(my-namespace/my-kafka-cluster)

   或者，直接编辑 ConfigMap:

   oc edit configmap strimzi-cluster-operator

2. 如果您更新了 YAML 文件而不是直接编辑 ConfigMap，请通过部署 ConfigMap 来应用更改：

   oc create -f install/cluster-operator/050-ConfigMap-strimzi-cluster-operator.yaml

流程

1. 创建 ConfigMap。

   您可以将 ConfigMap 作为 YAML 文件或属性文件创建。

   在本例中，过滤器属性仅为 my-topic 主题返回日志：

   kind: ConfigMap
   apiVersion: v1
   metadata:
     name: logging-configmap
   data:
     log4j2.properties:
       rootLogger.level="INFO"
       appender.console.filter.filter1.type=MarkerFilter
       appender.console.filter.filter1.onMatch=ACCEPT
       appender.console.filter.filter1.onMismatch=DENY
       appender.console.filter.filter1.marker=KafkaTopic(my-namespace/my-topic)

   如果您使用属性文件，请在命令行中指定该文件：

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

   属性文件定义日志配置：

   # Define the logger
   rootLogger.level="INFO"
   # Set the filters
   appender.console.filter.filter1.type=MarkerFilter
   appender.console.filter.filter1.onMatch=ACCEPT
   appender.console.filter.filter1.onMismatch=DENY
   appender.console.filter.filter1.marker=KafkaTopic(my-namespace/my-topic)
   # ...

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

   对于 Topic Operator，日志记录在 Kafka 资源的 topicOperator 配置中指定。

   spec:
     # ...
     entityOperator:
       topicOperator:
         logging:
           type: external
           valueFrom:
             configMapKeyRef:
               name: logging-configmap
               key: log4j2.properties

3. 通过部署 Cluster Operator 来应用更改：

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

流程

1. 配置 Kafka 资源。

   1. 配置 授权 属性。

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

      例如：

      apiVersion: kafka.strimzi.io/v1beta2
      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/v1beta2
   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/v1beta2
   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/v1beta2
   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。当您配置基于令牌的 oauth 身份验证时，您可以将 jwksEndpointUri 指定为本地 JWT 验证的 URI。tokenEndpointUri URI 的主机名必须相同。

   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 pod -w

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

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

1. 访问红帽单点登录管理控制台
2. 使用 Red Hat Single Sign-On 授权部署 Kafka 集群
3. 为 CLI Kafka 客户端会话准备 TLS 连接
4. 使用 CLI Kafka 客户端会话检查对 Kafka 的授权访问权限