在 OpenShift 中配置 AMQ Streams

流程

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

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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   oc apply -f <kafka_configuration_file>

流程

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

   例如：

   Copy to Clipboard Toggle word wrap

   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:

   Copy to Clipboard Toggle word wrap

   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:

   Copy to Clipboard Toggle word wrap

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

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

流程

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

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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   oc apply -f KAFKA-USER-CONFIG-FILE

流程

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

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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   oc apply -f <your-file>

流程

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

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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   oc apply -f MIRRORMAKER-CONFIGURATION-FILE

流程

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

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

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

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

   Copy to Clipboard Toggle word wrap

   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 集群配置示例

   Copy to Clipboard Toggle word wrap

   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 资源。

   Copy to Clipboard Toggle word wrap

   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 客户端身份验证的源用户配置示例

   Copy to Clipboard Toggle word wrap

   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 客户端身份验证的目标用户配置示例

   Copy to Clipboard Toggle word wrap

   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 资源。

   Copy to Clipboard Toggle word wrap

   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 配置示例

   Copy to Clipboard Toggle word wrap

   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 资源。

   Copy to Clipboard Toggle word wrap

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