2.2. Kafka 集群配置

流程

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       replicas: 3 1
       version: 3.4.0 2
       logging: 3
         type: inline
         loggers:
           kafka.root.logger.level: "INFO"
       resources: 4
         requests:
           memory: 64Gi
           cpu: "8"
         limits:
           memory: 64Gi
           cpu: "12"
       readinessProbe: 5
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       jvmOptions: 6
         -Xms: 8192m
         -Xmx: 8192m
       image: my-org/my-image:latest 7
       listeners: 8
         - name: plain 9
           port: 9092 10
           type: internal 11
           tls: false 12
           configuration:
             useServiceDnsDomain: true 13
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication: 14
             type: tls
         - name: external 15
           port: 9094
           type: route
           tls: true
           configuration:
             brokerCertChainAndKey: 16
               secretName: my-secret
               certificate: my-certificate.crt
               key: my-key.key
       authorization: 17
         type: simple
       config: 18
         auto.create.topics.enable: "false"
         offsets.topic.replication.factor: 3
         transaction.state.log.replication.factor: 3
         transaction.state.log.min.isr: 2
         default.replication.factor: 3
         min.insync.replicas: 2
         inter.broker.protocol.version: "3.4"
         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 日志记录器，您可以将日志级别设置为 INFO, ERROR, WARN, TRACE, DEBUG, FATAL 或 OFF。

   4

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

   5

   使用 HealthCheck 可以知道何时重启一个容器（存活度）以及何时一个容器可以开始接受流量（就绪度）。

   6

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

   7

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

   8

   侦听器配置客户端如何通过 bootstrap 地址连接到 Kafka 集群。监听器被配置为 internal 或 external 监听器，用于来自 OpenShift 集群内部或外的连接。

   9

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

   10

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

   11

   监听程序指定为 internal 或 cluster-ip (使用每个代理的 ClusterIP 服务公开 Kafka)，或为外部监听程序指定为 route（只限 OpenShift）, loadbalancer, nodeport 或 ingress（只限 Kubernetes）。

   12

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

   13

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

   14

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

   15

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

   16

   由外部 CA （证书授权）管理的 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 JMX Exporter 将指标导出到 Grafana 仪表板的 Prometheus 规则，通过引用包含 Prometheus JMX 导出器配置的 ConfigMap 来启用。您可以使用对在 metricsConfig.valueFrom.configMapKeyRef.key 下包含空文件的 ConfigMap 的引用来启用指标。

   26

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

   27

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

   28

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

   29

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

   30

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

   31

   指定的主题 Operator 日志程序和日志级别。这个示例使用 内联 日志记录。

   32

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

   33

   Kafka 导出程序配置。Kafka Exporter 是一个可选组件，用于在特定消费者滞后数据从 Kafka 代理中提取指标数据。要使 Kafka Exporter 正常工作，需要使用消费者组。

   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 > 以零开始，结束副本总数减一。例如，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 命名为 &lt ;cluster-name&gt; -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 被删除，然后重新创建时，等待下一个协调。