2.2. Kafka 集群配置

流程

1. 编辑集群的 Kafka 资源。

   更改 size 属性，以增大分配给 Kafka 集群的持久性卷的大小、一个 ZooKeeper 集群或两者。

   • 对于 Kafka 集群，更新 spec.kafka.storage 下的 size 属性。
   • 对于 ZooKeeper 集群，更新 spec.zookeeper.storage 下的 size 属性。

   将卷大小增加到 2000Gi的 Kafka 配置

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       storage:
         type: persistent-claim
         size: 2000Gi
         class: my-storage-class
       # ...
     zookeeper:
       # ...

   Copy to Clipboard Toggle word wrap

2. 创建或更新资源：

   oc apply -f <kafka_configuration_file>

   Copy to Clipboard Toggle word wrap

   OpenShift 增加所选持久性卷的容量，以响应 Cluster Operator 的请求。调整大小完成后，Cluster Operator 会重启所有使用调整大小的持久性卷的 pod。这会自动发生。

3. 验证集群中相关 pod 的存储容量是否已增加：

   oc get pv

   Copy to Clipboard Toggle word wrap

   带有更高存储的 Kafka 代理 pod

   NAME               CAPACITY   CLAIM
   pvc-0ca459ce-...   2000Gi     my-project/data-my-cluster-kafka-2
   pvc-6e1810be-...   2000Gi     my-project/data-my-cluster-kafka-0
   pvc-82dc78c9-...   2000Gi     my-project/data-my-cluster-kafka-1

   Copy to Clipboard Toggle word wrap

   输出显示与代理 pod 关联的每个 PVC 的名称。

流程

1. 编辑 Kafka 资源中的 spec.kafka.storage.volumes 属性。将新卷添加到 卷 阵列中。例如，添加新卷，其 ID 为 2：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
         - id: 1
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
         - id: 2
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
       # ...
     zookeeper:
       # ...

   Copy to Clipboard Toggle word wrap

2. 创建或更新资源：

   oc apply -f <kafka_configuration_file>

   Copy to Clipboard Toggle word wrap
3. 创建新主题或将现有分区重新分配给新磁盘。

流程

1. 从您要删除的磁盘中重新分配所有分区。分区中的任何数据仍会分配给要删除的磁盘中的数据可能会丢失。

2. 编辑 Kafka 资源中的 spec.kafka.storage.volumes 属性。从 volumes 阵列中删除一个或多个卷。例如，删除 ID 为 1 和 2 的卷：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
       # ...
     zookeeper:
       # ...

   Copy to Clipboard Toggle word wrap

3. 创建或更新资源：

   oc apply -f <kafka_configuration_file>

   Copy to Clipboard Toggle word wrap

流程

1. 从 Kafka 集群的 < cluster_name> -cluster-ca-cert secret 中提取集群 CA 证书和密钥。

   oc get secret <cluster_name>-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12

   Copy to Clipboard Toggle word wrap

   oc get secret <cluster_name>-cluster-ca-cert -o jsonpath='{.data.ca\.password}' | base64 -d > ca.password

   Copy to Clipboard Toggle word wrap

   将 <cluster_name > 替换为 Kafka 集群的名称。当您使用 Kafka 资源部署 Kafka 时，使用 Kafka 集群名称( <cluster_name> -cluster-ca-cert )创建一个带有集群 CA 证书的 secret。例如，my-cluster-cluster-ca-cert。

2. 使用 AMQ Streams Kafka 镜像运行一个新的交互式 pod 容器，以连接到正在运行的 Kafka 代理。

   oc run --restart=Never --image=registry.redhat.io/amq7/amq-streams-kafka-33-rhel8:2.3.0 <interactive_pod_name> -- /bin/sh -c "sleep 3600"

   Copy to Clipboard Toggle word wrap

   将 <interactive_pod_name > 替换为 pod 的名称。

3. 将集群 CA 证书复制到互动 pod 容器中。

   oc cp ca.p12 <interactive_pod_name>:/tmp

   Copy to Clipboard Toggle word wrap

4. 从具有访问 Kafka 代理的 Kafka 用户的 secret 中提取用户 CA 证书和密钥。

   oc get secret <kafka_user> -o jsonpath='{.data.user\.p12}' | base64 -d > user.p12

   Copy to Clipboard Toggle word wrap

   oc get secret <kafka_user> -o jsonpath='{.data.user\.password}' | base64 -d > user.password

   Copy to Clipboard Toggle word wrap

   将 <kafka_user > 替换为 Kafka 用户的名称。当使用 KafkaUser 资源创建 Kafka 用户时，会使用 Kafka 用户名创建带有用户 CA 证书的 secret。例如，my-user。

5. 将用户 CA 证书复制到交互式 pod 容器。

   oc cp user.p12 <interactive_pod_name>:/tmp

   Copy to Clipboard Toggle word wrap

   CA 证书允许交互式 pod 容器使用 TLS 连接到 Kafka 代理。

6. 创建 config.properties 文件，以指定用于验证与 Kafka 集群的连接的 truststore 和密钥存储。

   使用您在前面的步骤中提取的证书和密码。

   bootstrap.servers=<kafka_cluster_name>-kafka-bootstrap:9093 

   1

   
   security.protocol=SSL 

   2

   
   ssl.truststore.location=/tmp/ca.p12 

   3

   
   ssl.truststore.password=<truststore_password> 

   4

   
   ssl.keystore.location=/tmp/user.p12 

   5

   
   ssl.keystore.password=<keystore_password> 

   6

   Copy to Clipboard Toggle word wrap

   1

   连接到 Kafka 集群的 bootstrap 服务器地址。使用您自己的 Kafka 集群名称替换 < kafka_cluster_name>。

   2

   使用 TLS 进行加密时的安全协议选项。

   3

   truststore 位置包含了 Kafka 集群的公钥证书 (ca.p12)。

   4

   用于访问信任存储的密码(ca.password)。

   5

   密钥存储位置包含 Kafka 用户的公钥证书(user.p12)。

   6

   用于访问密钥存储的密码(user.password)。

7. 将 config.properties 文件复制到交互式 pod 容器中。

   oc cp config.properties <interactive_pod_name>:/tmp/config.properties

   Copy to Clipboard Toggle word wrap

8. 准备名为 topics.json 的 JSON 文件，以指定要移动的主题。

   将主题名称指定为用逗号分开的列表。

   用于重新分配 topic-a 和 topic-b的所有分区的 JSON 文件示例

   {
     "version": 1,
     "topics": [
       { "topic": "topic-a"},
       { "topic": "topic-b"}
     ]
   }

   Copy to Clipboard Toggle word wrap

9. 将 topics.json 文件复制到交互式 pod 容器中。

   oc cp topics.json <interactive_pod_name>:/tmp/topics.json

   Copy to Clipboard Toggle word wrap

10. 在互动 pod 容器中启动 shell 进程。

    oc exec -n <namespace> -ti <interactive_pod_name> /bin/bash

    Copy to Clipboard Toggle word wrap

    将 <namespace > 替换为运行 pod 的 OpenShift 命名空间。

11. 使用 kafka-reassign-partitions.sh 命令生成重新分配 JSON。

    示例以将 topic-a 和 topic-b 的所有分区移动到代理 0、1 和 2

    bin/kafka-reassign-partitions.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 \
      --command-config /tmp/config.properties \
      --topics-to-move-json-file /tmp/topics.json \
      --broker-list 0,1,2 \
      --generate

    Copy to Clipboard Toggle word wrap

流程

1. 通过增加 Kafka.spec.kafka.replicas 配置选项，根据需要添加多个新代理。
2. 验证新代理 pod 是否已启动。
3. 如果没有这样做，请运行交互式 pod 容器来生成一个重新分配 JSON 文件，名为 reassignment.json。

4. 将 reassignment.json 文件复制到交互式 pod 容器中。

   oc cp reassignment.json <interactive_pod_name>:/tmp/reassignment.json

   Copy to Clipboard Toggle word wrap

   将 <interactive_pod_name > 替换为 pod 的名称。

5. 在互动 pod 容器中启动 shell 进程。

   oc exec -n <namespace> -ti <interactive_pod_name> /bin/bash

   Copy to Clipboard Toggle word wrap

   将 <namespace > 替换为运行 pod 的 OpenShift 命名空间。

6. 使用交互式 pod 容器中的 kafka-reassign-partitions.sh 脚本来重新分配分区。

   bin/kafka-reassign-partitions.sh --bootstrap-server
    <cluster_name>-kafka-bootstrap:9093 \
    --command-config /tmp/config.properties \
    --reassignment-json-file /tmp/reassignment.json \
    --execute

   Copy to Clipboard Toggle word wrap

   将 <cluster_name > 替换为 Kafka 集群的名称。例如： my-cluster-kafka-bootstrap:9093

   如果要节流复制，也可以将 --throttle 选项传递给一个 inter-broker throttled 速率（以每秒字节为单位）。例如：

   bin/kafka-reassign-partitions.sh --bootstrap-server
     <cluster_name>-kafka-bootstrap:9093 \
     --command-config /tmp/config.properties \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 5000000 \
     --execute

   Copy to Clipboard Toggle word wrap

   此命令将输出两个重新分配 JSON 对象。第一个记录所移动分区的当前分配。如果稍后需要恢复重新分配，您应该将其保存到本地文件（而不是 pod 中的文件）。第二个 JSON 对象是您在重新分配 JSON 文件中传递的目标重新分配。

   如果您需要在重新分配时更改节流，您可以使用不同的节流率相同的命令。例如：

   bin/kafka-reassign-partitions.sh --bootstrap-server
     <cluster_name>-kafka-bootstrap:9093 \
     --command-config /tmp/config.properties \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 10000000 \
     --execute

   Copy to Clipboard Toggle word wrap

7. 验证重新分配已使用来自任何代理 pod 的 kafka-reassign-partitions.sh 命令行工具完成。这与上一步中的命令相同，但使用 --verify 选项而不是 --execute 选项。

   bin/kafka-reassign-partitions.sh --bootstrap-server
     <cluster_name>-kafka-bootstrap:9093 \
     --command-config /tmp/config.properties \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

   Copy to Clipboard Toggle word wrap

   当 --verify 命令报告每个分区都成功完成时，重新分配过程已被重新分配。最后 --verify 还会导致删除任何重新分配节流的效果。

8. 现在，如果保存了 JSON 以将分配恢复到其原始代理，您可以删除恢复的文件。

流程

1. 如果没有这样做，请运行交互式 pod 容器来生成一个重新分配 JSON 文件，名为 reassignment.json。

2. 将 reassignment.json 文件复制到交互式 pod 容器中。

   oc cp reassignment.json <interactive_pod_name>:/tmp/reassignment.json

   Copy to Clipboard Toggle word wrap

   将 <interactive_pod_name > 替换为 pod 的名称。

3. 在互动 pod 容器中启动 shell 进程。

   oc exec -n <namespace> -ti <interactive_pod_name> /bin/bash

   Copy to Clipboard Toggle word wrap

   将 <namespace > 替换为运行 pod 的 OpenShift 命名空间。

4. 使用交互式 pod 容器中的 kafka-reassign-partitions.sh 脚本来重新分配分区。

   bin/kafka-reassign-partitions.sh --bootstrap-server
    <cluster_name>-kafka-bootstrap:9093 \
    --command-config /tmp/config.properties \
    --reassignment-json-file /tmp/reassignment.json \
    --execute

   Copy to Clipboard Toggle word wrap

   将 <cluster_name > 替换为 Kafka 集群的名称。例如： my-cluster-kafka-bootstrap:9093

   如果要节流复制，也可以将 --throttle 选项传递给一个 inter-broker throttled 速率（以每秒字节为单位）。例如：

   bin/kafka-reassign-partitions.sh --bootstrap-server
     <cluster_name>-kafka-bootstrap:9093 \
     --command-config /tmp/config.properties \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 5000000 \
     --execute

   Copy to Clipboard Toggle word wrap

   此命令将输出两个重新分配 JSON 对象。第一个记录所移动分区的当前分配。如果稍后需要恢复重新分配，您应该将其保存到本地文件（而不是 pod 中的文件）。第二个 JSON 对象是您在重新分配 JSON 文件中传递的目标重新分配。

   如果您需要在重新分配时更改节流，您可以使用不同的节流率相同的命令。例如：

   bin/kafka-reassign-partitions.sh --bootstrap-server
     <cluster_name>-kafka-bootstrap:9093 \
     --command-config /tmp/config.properties \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 10000000 \
     --execute

   Copy to Clipboard Toggle word wrap

5. 验证重新分配已使用来自任何代理 pod 的 kafka-reassign-partitions.sh 命令行工具完成。这与上一步中的命令相同，但使用 --verify 选项而不是 --execute 选项。

   bin/kafka-reassign-partitions.sh --bootstrap-server
     <cluster_name>-kafka-bootstrap:9093 \
     --command-config /tmp/config.properties \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

   Copy to Clipboard Toggle word wrap

   当 --verify 命令报告每个分区都成功完成时，重新分配过程已被重新分配。最后 --verify 还会导致删除任何重新分配节流的效果。

6. 现在，如果保存了 JSON 以将分配恢复到其原始代理，您可以删除恢复的文件。

7. 当所有分区重新分配完成后，被删除的代理不应该负责集群中的任何分区。您可以通过检查代理的数据日志目录是否包含任何实时分区日志来验证这一点。如果代理上的日志目录包含一个与扩展的正则表达式 \.[a-z0-9]-delete$ 不匹配的目录，则代理仍然具有实时分区，且不应停止。

   您可以通过执行命令来检查这一点：

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     /bin/bash -c \
     "ls -l /var/lib/kafka/kafka-log_<n>_ | grep -E '^d' | grep -vE '[a-zA-Z0-9.-]+\.[a-z0-9]+-delete$'"

   Copy to Clipboard Toggle word wrap

   其中 n 是要删除的 pod 的数量。

   如果上述命令打印任何输出，则代理仍然具有实时分区。在这种情况下，重新分配过程没有完成，或者重新分配 JSON 文件不正确。

8. 当您确认代理没有实时分区时，您可以编辑 Kafka 资源的 Kafka.spec.kafka.replicas 属性来减少代理数量。

流程

1. 在 Kafka 资源中添加或编辑 maintenanceTimeWindows 属性。例如，允许 0800 到 1059 到 1400 到 1559 之间的维护，您可以设置 maintenanceTimeWindows，如下所示：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
     maintenanceTimeWindows:
       - "* * 8-10 * * ?"
       - "* * 14-15 * * ?"

   Copy to Clipboard Toggle word wrap

2. 创建或更新资源：

   oc apply -f <kafka_configuration_file>

   Copy to Clipboard Toggle word wrap