在 OpenShift 中配置 AMQ Streams

流程

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       replicas: 3 1
       version: 3.2.3 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
         default.replication.factor: 3
         min.insync.replicas: 2
         inter.broker.protocol.version: "3.2"
         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
       # ...

   1

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

   2

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

   3

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

   4

   用于保留 支持的资源、当前 cpu 和内存 的请求，以及指定可消耗的最大资源数量。

   5

   状况检查，了解何时重启容器（持续）以及容器是否可以接受流量（就绪状态）。

   6

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

   7

   ADVANCED OPTION： 容器镜像配置，这只在特殊情况下建议。

   8

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

   9

   用于标识监听程序的名称。必须在 Kafka 集群中唯一。

   10

   Kafka 内监听器使用的端口号。端口号必须在给定的 Kafka 集群中唯一。允许的端口号为 9092 及更高版本，除了端口 9404 和 9999 除外，后者已用于 Prometheus 和 JMX。根据监听程序类型，端口号可能与连接 Kafka 客户端的端口号不同。

   11

   监听程序类型指定为 internal，或者如果为外部监听程序，指定为 route, loadbalancer, nodeport 或 ingress。

   12

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

   13

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

   14

   指定为 mutual TLS、SCRAM-SHA-512 或基于令牌的 OAuth 2.0 的监听程序验证机制。

   15

   外部监听程序配置指定 Kafka 集群如何在 OpenShift 以外公开，如通过 route, loadbalancer 或 nodeport。

   16

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

   17

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

   18

   代理配置。可以提供标准 Apache Kafka 配置，仅限于不直接由 AMQ Streams 管理的属性。

   19

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

   20

   存储配置为 临时、persistent-claim 或 jbod。

   21

   可能会增加 持久性卷的存储大小，也可以将 额外的 卷添加到 JBOD 存储中。

   22

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

   23

   机架感知 配置，将副本分散到不同的机架、数据中心或可用性区域。topologyKey 必须与包含机架 ID 的节点标签匹配。此配置中使用的示例使用标准 topology.kubernetes.io/zone 标签指定区。

   24

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

   25

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

   26

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

   27

   ZooKeeper 节点数量。ZooKeeper 集群通常有奇数个节点，一般为三个、五个或七个。大多数节点都必须可用才能保持有效的仲裁。如果 ZooKeeper 集群丢失了仲裁，它将停止对客户端进行响应，并且 Kafka 代理将停止工作。拥有稳定且高度可用的 ZooKeeper 集群对于 AMQ Streams 至关重要。

   28

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

   29

   实体 Operator 配置，指定主题 Operator 和用户 Operator 的配置。

   30

   实体 Operator TLS sidecar 配置。实体 Operator 使用 TLS sidecar 进行与 ZooKeeper 的安全通信。

   31

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

   32

   指定 User Operator 日志记录器和日志级别。

   33

   Kafka 导出器配置.Kafka Exporter 是一个可选组件，用于从 Kafka 代理中提取指标数据。

   34

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

2. 创建或更新资源：

   oc apply -f <kafka_configuration_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 的名称。

   Kafka 代理 pod 的名称为 & lt;cluster-name&gt; -kafka-<index & gt;，其中 <index> 从零开始，结尾是 replicas减一总数。例如，my-cluster-kafka-0。

2. 注解 OpenShift 中的 Pod 资源。

   使用 oc annotate:

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

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

流程

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

   zookeeper pod 的名称为 < cluster-name>-zookeeper-<index &gt ;，其中 <index > 从零开始，结尾是 replicas minus 总数。例如，my-cluster-zookeeper-0。

2. 注解 OpenShift 中的 Pod 资源。

   使用 oc annotate:

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

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

流程

1. 编辑 KafkaConnect 资源的 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。

   2

   为 Kafka Connect 集群启用 KafkaConnectors。

   3

   运行任务的 worker 的副本数。

   4

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

   5

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

   6

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

   7

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

   8

   构建配置属性，以使用连接器插件构建容器镜像。

   9

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

   10

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

   11

   Kafka 连接器的外部配置，如此处所示或卷。您还可以使用 配置供应商插件 从外部源 加载配置值。

   12

   用于保留 支持的资源、当前 cpu 和内存 的请求，以及指定可消耗的最大资源数量。

   13

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

   14

   状况检查，了解何时重启容器（持续）以及容器是否可以接受流量（就绪状态）。

   15

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

   16

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

   17

   ADVANCED OPTION： 容器镜像配置，这只在特殊情况下建议。

   18

   SPECIALIZED OPTION ：对部署的意识 配置.这是用于在同一位置（而非跨地区）部署的专用选项。如果您希望连接器从最接近的副本而不是领导副本使用，则使用此选项。在某些情况下，消耗自最接近的副本可以提高网络利用率或降低成本。topologyKey 必须与包含机架 ID 的节点标签匹配。此配置中使用的示例使用标准 topology.kubernetes.io/zone 标签指定区。要从最接近的副本使用，请在 Kafka 代理配置中启用 RackAwareReplicaSelector。

   19

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

   20

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

2. 创建或更新资源：

   oc apply -f KAFKA-CONNECT-CONFIG-FILE

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

流程

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

   在以下示例中，使用 字面 名称值为 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. 编辑 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

   面向消费者和生成器的 bootstrap 服务器。

   3

   使用者的组 ID。

   4

   消费者流的数量。

   5

   offset auto-commit 间隔（以毫秒为单位）。

   6

   TLS 加密，使用密钥名称，其中 TLS 证书存储为消费者或制作者的 X.509 格式。如果证书存储在同一 secret 中，可以多次列出它。

   7

   使用 TLS 机制的用于消费者或生成者的验证（如此处所示），使用 OAuth bearer tokens, 或一个基于 SASL 的 SCRAM-SHA-256/SCRAM-SHA-512 或 PLAIN 机制。

   8

   consumer 和 producer 的 Kafka 配置选项。

   9

   使用 TLS 版本的特定 密码套件 运行外部监听程序的 SSL 属性。

   10

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

   11

   如果 abortOnSendFailure 属性设为 true，则 Kafka MirrorMaker 将退出，并且容器将因消息的发送失败而重启。

   12

   使用 TLS 版本的特定 密码套件 运行外部监听程序的 SSL 属性。

   13

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

   14

   包含从源到目标 Kafka 集群的主题。

   15

   用于保留 支持的资源、当前 cpu 和内存 的请求，以及指定可消耗的最大资源数量。

   16

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

   17

   状况检查，了解何时重启容器（持续）以及容器是否可以接受流量（就绪状态）。

   18

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

   19

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

   20

   ADVANCED OPTION： 容器镜像配置，这只在特殊情况下建议。

   21

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

   22

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

   23

   为 Jaeger 启用分布式追踪。

   警告

   将 abortOnSendFailure 属性设置为 false 时，生产者会尝试在主题中发送下一消息。原始消息可能会丢失，因为没有尝试重新发送失败的信息。

2. 创建或更新资源：

   oc apply -f <your-file>

流程

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker2
   spec:
     version: 3.2.3 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: "org.apache.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: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
       topicsPattern: "topic1|topic2|topic3" 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
     rack:
       topologyKey: topology.kubernetes.io/zone 42
     template: 43
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 44
         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 45
     externalConfiguration: 46
       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

   运行任务的 worker 的副本数。

   3

   Kafka Connect 的 Kafka 集群别名，必须指定 目标 Kafka 集群。Kafka 连接使用 Kafka 集群作为其内部主题。

   4

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

   5

   源 Kafka 集群的集群别名。

   6

   源集群的身份验证使用 TLS 机制，如此处所示，使用 OAuth bearer 令牌或基于 SASL 的 SCRAM-SHA-256/SCRAM-SHA-512 或 PLAIN 机制。

   7

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

   8

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

   9

   目标 Kafka 集群的集群别名。

   10

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

   11

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

   12

   Kafka 连接配置.可以提供标准 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 在集群的成员之间分发任务。如果任务数量超过 worker，则 worker 会被分配多个任务。对于接收器连接器，旨在为每个消耗的主题分区有一个任务。对于源连接器，可以并行运行的任务数量也可能依赖于外部系统。如果无法实现并行，连接器会创建一个数量少于最大任务数量。

   21

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

   22

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

   23

   启用 ACL 规则同步后，ACL 将应用到同步主题。默认值是 true。此功能与 User Operator 不兼容。如果使用 User Operator，请将此属性设置为 false。

   24

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

   25

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

   26

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

   27

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

   28

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

   29

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

   30

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

   31

   可选设置，用于更改新消费者组的检查频率。默认值每 10 分钟进行一次检查。

   32

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

   33

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

   34

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

   35

   来自源集群的主题复制 ，定义为用逗号分开的列表或正则表达式模式。源连接器复制指定的主题。Checkpoint 连接器跟踪指定主题的偏移量。此处我们按名称请求三个主题。

   36

   来自源集群的消费者组复制 ，定义为以逗号分隔的列表或正则表达式模式。checkpoint 连接器复制指定的使用者组。此处我们根据名称请求三个消费者组。

   37

   用于保留 支持的资源、当前 cpu 和内存 的请求，以及指定可消耗的最大资源数量。

   38

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

   39

   状况检查，了解何时重启容器（持续）以及容器是否可以接受流量（就绪状态）。

   40

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

   41

   ADVANCED OPTION： 容器镜像配置，这只在特殊情况下建议。

   42

   SPECIALIZED OPTION ：对部署的意识 配置.这是用于在同一位置（而非跨地区）部署的专用选项。如果您希望连接器从最接近的副本而不是领导副本使用，则使用此选项。在某些情况下，消耗自最接近的副本可以提高网络利用率或降低成本。topologyKey 必须与包含机架 ID 的节点标签匹配。此配置中使用的示例使用标准 topology.kubernetes.io/zone 标签指定区。要从最接近的副本使用，请在 Kafka 代理配置中启用 RackAwareReplicaSelector。

   43

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

   44

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

   45

   为 Jaeger 启用分布式追踪。

   46

   OpenShift Secret 的外部配置 作为环境变量挂载到 Kafka MirrorMaker。您还可以使用 配置供应商插件 从外部源 加载配置值。

2. 创建或更新资源：

   oc apply -f MIRRORMAKER-CONFIGURATION-FILE

流程

1. 配置两个 Kafka 资源，一个用于保护源 Kafka 集群，另一个来保护目标 Kafka 集群。

   您可以添加用于身份验证的监听程序配置并启用授权。

   在本例中，为使用 TLS 加密和身份验证的 Kafka 集群配置内部监听程序。启用 Kafka 简单 授权。

   使用 TLS 身份验证的源 Kafka 集群配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-source-cluster
   spec:
     kafka:
       version: 3.2.3
       replicas: 1
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
       authorization:
         type: simple
       config:
         offsets.topic.replication.factor: 1
         transaction.state.log.replication.factor: 1
         transaction.state.log.min.isr: 1
         default.replication.factor: 1
         min.insync.replicas: 1
         inter.broker.protocol.version: "3.2"
       storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
     zookeeper:
       replicas: 1
       storage:
         type: persistent-claim
         size: 100Gi
         deleteClaim: false
     entityOperator:
       topicOperator: {}
       userOperator: {}

   使用 TLS 身份验证的目标 Kafka 集群配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-target-cluster
   spec:
     kafka:
       version: 3.2.3
       replicas: 1
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
       authorization:
         type: simple
       config:
         offsets.topic.replication.factor: 1
         transaction.state.log.replication.factor: 1
         transaction.state.log.min.isr: 1
         default.replication.factor: 1
         min.insync.replicas: 1
         inter.broker.protocol.version: "3.2"
       storage:
         type: jbod
         volumes:
           - id: 0
             type: persistent-claim
             size: 100Gi
             deleteClaim: false
     zookeeper:
       replicas: 1
       storage:
         type: persistent-claim
         size: 100Gi
         deleteClaim: false
     entityOperator:
       topicOperator: {}
       userOperator: {}

2. 在单独的命名空间中创建或更新 Kafka 资源。

   oc apply -f <kafka_configuration_file> -n <namespace>

   Cluster Operator 创建监听程序并设置集群和客户端证书颁发机构(CA)证书，以便在 Kafka 集群中启用身份验证。

   证书在 secret < cluster_name> -cluster-ca-cert 中创建。

3. 配置两个 KafkaUser 资源，一个用于源 Kafka 集群的用户，一个用于目标 Kafka 集群的用户。

   1. 配置与对应的源和目标 Kafka 集群相同的身份验证和授权类型。例如，如果您在源 Kafka 集群的 Kafka 配置中使用了 tls 验证和 simple 授权类型，请在 KafkaUser 配置中使用相同的。

   2. 配置 MirrorMaker 2.0 所需的 ACL，以允许在源和目标 Kafka 集群上执行操作。

      ACL 由内部 MirrorMaker 连接器以及底层 Kafka 连接框架使用。

   TLS 客户端身份验证的源用户配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-source-user
     labels:
       strimzi.io/cluster: my-source-cluster
   spec:
     authentication:
       type: tls
     authorization:
       type: simple
       acls:
         # MirrorSourceConnector
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operation: Create
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operation: DescribeConfigs
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operation: Write
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Read
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: DescribeConfigs
         # MirrorCheckpointConnector
         - resource:
             type: cluster
           operation: Describe
         - resource: # Needed for every group for which offsets are synced
             type: group
             name: "*"
           operation: Describe
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operation: Read

   TLS 客户端身份验证的目标用户配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-target-user
     labels:
       strimzi.io/cluster: my-target-cluster
   spec:
     authentication:
       type: tls
     authorization:
       type: simple
       acls:
         # Underlying Kafka Connect internal topics to store configuration, offsets, or status
         - resource:
             type: group
             name: mirrormaker2-cluster
           operation: Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: Describe
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: DescribeConfigs
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: Create
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: Describe
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: DescribeConfigs
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: Create
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: Describe
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: DescribeConfigs
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: Create
         # MirrorSourceConnector
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Create
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Alter
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: AlterConfigs
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Write
         # MirrorCheckpointConnector
         - resource:
             type: cluster
           operation: Describe
         - resource:
             type: topic
             name: my-source-cluster.checkpoints.internal
           operation: Create
         - resource:
             type: topic
             name: my-source-cluster.checkpoints.internal
           operation: Describe
         - resource:
             type: topic
             name: my-source-cluster.checkpoints.internal
           operation: Write
         - resource: # Needed for every group for which the offset is synced
             type: group
             name: "*"
           operation: Read
         - resource: # Needed for every group for which the offset is synced
             type: group
             name: "*"
           operation: Describe
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Read
         # MirrorHeartbeatConnector
         - resource:
             type: topic
             name: heartbeats
           operation: Create
         - resource:
             type: topic
             name: heartbeats
           operation: Describe
         - resource:
             type: topic
             name: heartbeats
           operation: Write

   注意

   您可以通过将 type 设置为 tls-external 来使用 User Operator 外部发布的证书。如需更多信息，请参阅 用户身份验证。

4. 在您为源和目标集群创建的每个命名空间中创建或更新 KafkaUser 资源。

   oc apply -f <kafka_user_configuration_file> -n <namespace>

   User Operator 根据所选的身份验证类型创建代表客户端(MirrorMaker)和用于客户端身份验证的安全凭证的用户。

   User Operator 会创建一个与 KafkaUser 资源名称相同的新 secret。secret 包含用于 TLS 客户端身份验证的私钥和公钥。公钥包含在用户证书中，由客户端证书颁发机构(CA)签名。

5. 使用身份验证详情配置 KafkaMirrorMaker2 资源，以连接到源和目标集群。

   带有 TLS 身份验证的 MirrorMaker 2.0 配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker-2
   spec:
     version: 3.2.3
     replicas: 1
     connectCluster: "my-target-cluster"
     clusters:
       - alias: "my-source-cluster"
         bootstrapServers: my-source-cluster-kafka-bootstrap:9093
         tls: 1
           trustedCertificates:
             - secretName: my-source-cluster-cluster-ca-cert
               certificate: ca.crt
         authentication: 2
           type: tls
           certificateAndKey:
             secretName: my-source-user
             certificate: user.crt
             key: user.key
       - alias: "my-target-cluster"
         bootstrapServers: my-target-cluster-kafka-bootstrap:9093
         tls: 3
           trustedCertificates:
             - secretName: my-target-cluster-cluster-ca-cert
               certificate: ca.crt
         authentication: 4
           type: tls
           certificateAndKey:
             secretName: my-target-user
             certificate: user.crt
             key: user.key
         config:
           # -1 means it will use the default replication factor configured in the broker
           config.storage.replication.factor: -1
           offset.storage.replication.factor: -1
           status.storage.replication.factor: -1
     mirrors:
       - sourceCluster: "my-source-cluster"
         targetCluster: "my-target-cluster"
         sourceConnector:
           config:
             replication.factor: 1
             offset-syncs.topic.replication.factor: 1
             sync.topic.acls.enabled: "false"
         heartbeatConnector:
           config:
             heartbeats.topic.replication.factor: 1
         checkpointConnector:
           config:
             checkpoints.topic.replication.factor: 1
             sync.group.offsets.enabled: "true"
         topicsPattern: "topic1|topic2|topic3"
         groupsPattern: "group1|group2|group3"

   1

   源 Kafka 集群的 TLS 证书。如果它们位于单独的命名空间中，请从 Kafka 集群的命名空间复制集群 secret。

   2

   使用 TLS 机制 访问源 Kafka 集群的用户身份验证。

   3

   目标 Kafka 集群的 TLS 证书。

   4

   访问目标 Kafka 集群的用户身份验证。

6. 在与目标 Kafka 集群相同的命名空间中创建或更新 KafkaMirrorMaker2 资源。

   oc apply -f <mirrormaker2_configuration_file> -n <namespace_of_target_cluster>

流程

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

   oc get KafkaMirrorMaker2

2. 查找 Kafka MirrorMaker 2.0 连接器的名称，以便从 KafkaMirrorMaker2 自定义资源重启。

   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 重启名为 my-source->my-target.MirrorSourceConnector 的连接器的任务 0 ：

   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: <cluster_name>-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 服务器。使用 Kafka 集群的名称作为 < cluster_name>。

   3

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

   4

   Kafka Bridge 集群的身份验证，使用 TLS 机制，如此处所示，使用 OAuth bearer 令牌 或基于 SASL 的 SCRAM-SHA-256/SCRAM-SHA-512 或 PLAIN 机制。默认情况下，Kafka Bridge 会在没有身份验证的情况下连接到 Kafka 代理。

   5

   对 Kafka 代理的 HTTP 访问。

   6

   CORS 访问 指定所选资源和访问方法。请求中的其他 HTTP 标头描述了允许访问 Kafka 集群的来源。

   7

   使用者配置选项。

   8

   制作者配置选项。

   9

   用于保留 支持的资源、当前 cpu 和内存 的请求，以及指定可消耗的最大资源数量。

   10

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

   11

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

   12

   状况检查，了解何时重启容器（持续）以及容器是否可以接受流量（就绪状态）。

   13

   可选： 容器镜像配置，这只在特殊情况下建议。

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

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

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

3. 创建或更新资源。

   oc apply -f <kafka_configuration_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_configuration_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_configuration_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_configuration_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 的名称，并将 logging.valueFrom.configMapKeyRef.key 设置为此 ConfigMap 中的键。

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

3. 创建或更新资源。

   oc apply -f <kafka_configuration_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 的名称，并将 logging.valueFrom.configMapKeyRef.key 设置为此 ConfigMap 中的键。

   对于主题 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 令牌的身份验证
4. 自定义身份验证

流程

1. 配置 Kafka 资源。

   1. 配置 授权 属性以进行授权。

   2. 配置 监听程序 属性，以创建具有身份验证的监听程序。

      例如：

      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_configuration_file>

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

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

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

   在 secret <cluster _name> -cluster-ca-cert 中也创建了用于验证 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 或 scram-sha-512。

   2

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

2. 创建或更新 KafkaUser 资源。

   oc apply -f <user_config_file>

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

流程

1. 打开 Kafka 资源。

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

   例如，要配置 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 Streams 集成
2. 使用配置为使用 OAuth 2.0 的 Kafka 代理监听程序部署或更新 Kafka 集群
3. 更新基于 Java 的 Kafka 客户端以使用 OAuth 2.0
4. 更新 Kafka 组件客户端以使用 OAuth 2.0