第 2 章 部署配置

步骤

1. 在 Kafka 资源中，增大分配给 Kafka 集群和 ZooKeeper 集群的持久性卷的大小。

   • 要增大分配给 Kafka 集群的卷大小，编辑 spec.kafka.storage 属性。

   • 要增大分配给 ZooKeeper 集群的卷大小，请编辑 spec.zookeeper.storage 属性。

     例如，将卷大小从 1000Gi 增加到 2000Gi ：

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

2. 创建或更新资源。

   使用 oc apply:

   oc apply -f your-file

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

步骤

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

   apiVersion: kafka.strimzi.io/v1beta1
   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:
       # ...

2. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f KAFKA-CONFIG-FILE

3. 创建新主题或将现有分区重新分配至新磁盘.

步骤

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

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

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

3. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f your-file

步骤

1. 编辑 Kafka 资源中的 replicas 属性。例如：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       replicas: 3
       # ...
     zookeeper:
       # ...

2. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f your-file

步骤

1. 打开包含指定集群部署的 Kafka 资源的 YAML 配置文件。

2. 在 Kafka 资源中的 spec.kafka.config 属性中，输入一个或多个 Kafka 配置设置。例如：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       config:
         default.replication.factor: 3
         offsets.topic.replication.factor: 3
         transaction.state.log.replication.factor: 3
         transaction.state.log.min.isr: 1
       # ...
     zookeeper:
       # ...

3. 应用新配置以创建或更新资源。

   使用 oc apply:

   oc apply -f kafka.yaml

   其中 kafka.yaml 是您要配置的资源的 YAML 配置文件，例如 kafka-persistent.yaml。

步骤

1. 打开包含指定集群部署的 Kafka 资源的 YAML 配置文件。

2. 在 Kafka 资源中的 spec.zookeeper.replicas 属性中，输入复制的 ZooKeeper 服务器数量。例如：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       replicas: 3
       # ...

3. 应用新配置以创建或更新资源。

   使用 oc apply:

   oc apply -f kafka.yaml

   其中 kafka.yaml 是您要配置的资源的 YAML 配置文件，例如 kafka-persistent.yaml。

步骤

1. 打开包含指定集群部署的 Kafka 资源的 YAML 配置文件。

2. 在 Kafka 资源中的 spec.zookeeper.config 属性中，输入一个或多个 ZooKeeper 配置设置。例如：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       config:
         autopurge.snapRetainCount: 3
         autopurge.purgeInterval: 1
       # ...

3. 应用新配置以创建或更新资源。

   使用 oc apply:

   oc apply -f kafka.yaml

   其中 kafka.yaml 是您要配置的资源的 YAML 配置文件，例如 kafka-persistent.yaml。

步骤

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

   例如：

   oc exec -it my-cluster-zookeeper-0 -- bin/kafka-topics.sh --list --zookeeper localhost:12181

   务必使用 localhost:12181。

   现在，您可以向 ZooKeeper 运行 Kafka 命令。

步骤

1. 编辑 Kafka 资源中的 principal Operator 属性。例如：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
     entityOperator:
       topicOperator:
         watchedNamespace: my-topic-namespace
         reconciliationIntervalSeconds: 60
       userOperator:
         watchedNamespace: my-user-namespace
         reconciliationIntervalSeconds: 60

2. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f your-file

步骤

1. 编辑指定集群部署的资源中的 resources 属性。例如：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       resources:
         requests:
           cpu: "8"
           memory: 64Gi
         limits:
           cpu: "12"
           memory: 128Gi
       # ...
     zookeeper:
       # ...

2. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f your-file

步骤

1. 请咨询您的 OpenShift 管理员，了解代表节点要部署到的区域 / 机架的节点标签。

2. 将标签用作拓扑键，编辑 Kafka 资源中的 rack 属性。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       rack:
         topologyKey: topology.kubernetes.io/zone
       # ...

3. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f your-file

步骤

1. 编辑 Kafka 资源中的 liveness Probe 或 readinessProbe 属性。例如：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       readinessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       # ...
     zookeeper:
       # ...

2. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f your-file

步骤

1. 编辑 Kafka 资源中的 metrics 属性。例如：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       metrics:
         lowercaseOutputName: true
       # ...

2. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f your-file

步骤

1. 编辑 Kafka、KafkaConnect、Kafka Connect S2I、Kafka MirrorMaker 或 Kafka Bridge 资源中的 jvmOptions 属性。例如：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       jvmOptions:
         "-Xmx": "8g"
         "-Xms": "8g"
       # ...
     zookeeper:
       # ...

2. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f your-file

步骤

1. 编辑 Kafka 资源中的 image 属性。例如：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       image: my-org/my-image:latest
       # ...
     zookeeper:
       # ...

2. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f your-file

步骤

1. 编辑 Kafka 资源中的 tlsSidecar 属性。例如：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
     entityOperator:
       # ...
       tlsSidecar:
         resources:
           requests:
             cpu: 200m
             memory: 64Mi
           limits:
             cpu: 500m
             memory: 128Mi
       # ...
     cruiseControl:
       # ...
       tlsSidecar:
         resources:
           requests:
             cpu: 200m
             memory: 64Mi
           limits:
             cpu: 500m
             memory: 128Mi
       # ...

2. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f your-file

步骤

1. 准备名为 topic .json 的 JSON 文件，其中列出要移动的主题。它必须具有以下结构：

   {
     "version": 1,
     "topics": [
       <TopicObjects>
     ]
   }

   其中 <TopicObjects> 是一个以逗号分隔的对象列表，例如：

   {
     "topic": <TopicName>
   }

   例如，如果要重新分配 topic-a 和 topic- b 的所有 分区，则需要准备类似以下的 topic .json 文件：

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

2. 将 topic.json 文件复制到其中一个代理 pod 中：

   cat topics.json | oc exec -c kafka <BrokerPod> -i -- \
     /bin/bash -c \
     'cat > /tmp/topics.json'

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

   oc exec <BrokerPod> -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --topics-to-move-json-file /tmp/topics.json \
     --broker-list <BrokerList> \
     --generate

   例如，将 topic-a 和 topic- b 的所有分区移动到 broker 4 和 7

   oc exec <BrokerPod> -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --topics-to-move-json-file /tmp/topics.json \
     --broker-list 4,7 \
     --generate

步骤

1. 通过增加 Kafka.spec.kafka.replicas 配置选项，根据需要添加多个新代理。
2. 验证新代理 Pod 是否已启动。

3. 将 reassignment.json 文件复制到您稍后要执行命令的代理 pod 中：

   cat reassignment.json | \
     oc exec broker-pod -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

   例如：

   cat reassignment.json | \
     oc exec my-cluster-kafka-0 -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

4. 使用同一代理 pod 的 kafka-reassign-partitions.sh 命令行工具执行分区重新分配。

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --execute

   如果您要节流复制，您也可以使用 --throttle 选项，每秒使用节流率（以字节为单位）。例如：

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 5000000 \
     --execute

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

5. 如果您需要在重新分配期间更改节流，您可以使用具有不同节流率的同一命令行。例如：

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 10000000 \
     --execute

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

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

   例如，

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

7. 当 --verify 命令报告每个分区成功完成时，重新分配已完成。最后的 --verify 还会导致移除任何重新分配节流。现在，如果您保存了 JSON，将分配还原到其原始代理中，您可以删除还原文件。

步骤

1. 将 reassignment.json 文件复制到您稍后要执行命令的代理 pod 中：

   cat reassignment.json | \
     oc exec broker-pod -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

   例如：

   cat reassignment.json | \
     oc exec my-cluster-kafka-0 -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

2. 使用同一代理 pod 的 kafka-reassign-partitions.sh 命令行工具执行分区重新分配。

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --execute

   如果您要节流复制，您也可以使用 --throttle 选项，每秒使用节流率（以字节为单位）。例如：

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 5000000 \
     --execute

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

3. 如果您需要在重新分配期间更改节流，您可以使用具有不同节流率的同一命令行。例如：

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 10000000 \
     --execute

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

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

   例如，

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

5. 当 --verify 命令报告每个分区成功完成时，重新分配已完成。最后的 --verify 还会导致移除任何重新分配节流。现在，如果您保存了 JSON，将分配还原到其原始代理中，您可以删除还原文件。

6. 所有分区重新分配完成后，删除的代理不应对集群中的任何分区负责。您可以通过检查代理的数据日志目录是否包含任何实时分区日志来验证这一点。如果代理上的日志目录包含与扩展正则表达式 [a-zA-Z0-9.-]+\ 不匹配的目录。[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$'"

   其中 N 是要删除的 Pod 数。

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

7. 确认代理没有实时分区后，您可以编辑 Kafka.spec.kafka.replicas 资源的 Kafka. spec.kafka.replicas，这会缩减 StatefulSet，从而删除最高数字的代理 Pod。

步骤

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

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

2. 创建或更新资源。

   这可以使用 oc apply 来完成：

   oc apply -f your-file