20.2. 生成重新分配 JSON 文件来重新分配分区

流程

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

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

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

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

   oc run --restart=Never --image=registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0 <interactive_pod_name> -- /bin/sh -c "sleep 3600"

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

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

   oc cp ca.p12 <interactive_pod_name>:/tmp

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

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

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

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

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

   oc cp user.p12 <interactive_pod_name>:/tmp

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

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

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

   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

   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

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

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

   重新分配 my-topic的所有分区的 JSON 文件示例

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

   您还可以使用此文件 更改主题 的复制因素。

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

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

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

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

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

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

    将 my-topic 分区移到指定代理的示例

    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,3,4 \
      --generate