第 7 章 升级 AMQ Streams

步骤

1. 记录对现有 Cluster Operator 资源所做的任何配置更改（在 /install/cluster-operator 目录中）。任何更改都将被 Cluster Operator 的新版本 覆盖。

2. 更新 Cluster Operator。

   1. 根据 Cluster Operator 运行的命名空间修改新版本的安装文件。

      在 Linux 中，使用：

      sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

      Copy to Clipboard

      在 MacOS 中，使用：

      sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

      Copy to Clipboard
   2. 如果在现有 Cluster Operator Deployment 中修改了一个或多个环境变量，请编辑 install/cluster-operator/060-Deployment-cluster-operator.yaml 文件来使用这些环境变量。

3. 当您有更新的配置时，请将其与其它安装资源一起部署：

   oc apply -f install/cluster-operator

   Copy to Clipboard

   等待滚动更新完成。

4. 获取 Kafka pod 的镜像以确保升级成功：

   oc get po my-cluster-kafka-0 -o jsonpath='{.spec.containers[0].image}'

   Copy to Clipboard

   镜像标签显示新的 AMQ Streams 版本，后跟 Kafka 版本。例如，<New AMQ Streams version>-kafka-<Current Kafka version>。

5. 更新现有资源以处理已弃用的自定义资源属性。

   • AMQ Streams 资源升级

步骤

1. 更新 Kafka 集群配置：

   oc edit kafka my-cluster

   Copy to Clipboard

2. 如果配置了，请确保 Kafka.spec.kafka.config 的 log.message.format .version 和 inter.broker.protocol.version 设置为 当前 Kafka 版本的默认值。

   例如，如果从 Kafka 版本 2.5.0 升级到 2.6.0：

   kind: Kafka
   spec:
     # ...
     kafka:
       version: 2.5.0
       config:
         log.message.format.version: "2.5"
         inter.broker.protocol.version: "2.5"
         # ...

   Copy to Clipboard

   如果没有配置 log.message.format .version 和 inter.broker.protocol.version，AMQ Streams 会在下一步中将这些版本更新至 Kafka 版本后自动将这些版本更新至当前的默认值。

   注意

   log.message.format.version 和 inter.broker.protocol.version 的值必须是字符串，以防止它们被解释为浮动点数。

3. 更改 Kafka.spec.kafka.version 以指定新的 Kafka 版本；保留 log.message.format.version 和 inter.broker.protocol.version，默认为 当前 Kafka 版本。

   注意

   更改 kafka.version 可确保集群中的所有代理都将升级为使用新的代理二进制文件。在此过程中，一些代理使用旧二进制文件，另一些则已升级至新二进制文件。不要更改 inter.broker.protocol.version，确保代理可以在升级过程中继续相互通信。

   例如，如果从 Kafka 2.5.0 升级到 2.6.0：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 2.6.0 

   1

   
       config:
         log.message.format.version: "2.5" 

   2

   
         inter.broker.protocol.version: "2.5" 

   3

   
         # ...

   Copy to Clipboard

   1

   Kafka 版本已改为新版本。

   2

   消息格式版本保持不变。

   3

   代理协议版本没有改变。

   警告

   如果新 Kafka 版本的 inter.broker.protocol.version 发生变化，则无法降级 Kafka。broker 协议版本决定了代理存储的持久元数据的模式，包括写入到 __consumer_offset 的消息。降级集群将不会理解消息。

4. 如果在 Kafka 自定义资源（在 Kafka .spec.kafka.image 中）中定义了 Kafka 集群 的镜像，请将镜像 更新为指向具有新 Kafka 版本的容器镜像。

   请参阅 Kafka 版本和镜像映射

5. 保存并退出编辑器，然后等待滚动更新完成。

   通过观察 pod 状态转换来检查滚动更新的进度：

   oc get pods my-cluster-kafka-0 -o jsonpath='{.spec.containers[0].image}'

   Copy to Clipboard

   滚动更新可确保每个 pod 在新版本的 Kafka 中使用代理二进制文件。

6. 根据您选择的 升级客户端策略，升级所有客户端应用以使用新版本的客户端二进制文件。

   如果需要，将 Kafka Connect 和 MirrorMaker 的 version 属性设置为 Kafka 的新版本：

   1. 对于 Kafka Connect，更新 KafkaConnect.spec.version。
   2. 对于 MirrorMaker，更新 KafkaMirrorMaker.spec.version。
   3. 对于 MirrorMaker 2.0，更新 KafkaMirrorMaker2.spec.version。

7. 如果配置，更新 Kafka 资源，以使用新的 inter.broker.protocol.version 版本。否则，请转至第 9 步。

   例如，如果升级到 Kafka 2.6.0：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 2.6.0
       config:
         log.message.format.version: "2.5"
         inter.broker.protocol.version: "2.6"
         # ...

   Copy to Clipboard
8. 等待 Cluster Operator 更新集群。

9. 如果配置，更新 Kafka 资源，以使用新的 log.message.format.version 版本。否则，请转至第 10 步。

   例如，如果升级到 Kafka 2.6.0：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 2.6.0
       config:
         log.message.format.version: "2.6"
         inter.broker.protocol.version: "2.6"
         # ...

   Copy to Clipboard

10. 等待 Cluster Operator 更新集群。

    • Kafka 集群和客户端现在使用新的 Kafka 版本。
    • 代理配置为使用 Kafka 的新版本的 Inter-broker 协议版本和消息格式发送消息。

1. 将 Kafka client .jar 文件替换为新版本。
2. 在使用者配置中，将 合作粘滞 附加到 partition.assignment.strategy。例如，如果设定了 范围 策略，请将配置更改为 范围，协作粘性。
3. 依次重新启动组中的每个使用者，等待使用者在每次重启后重新加入组。
4. 通过从使用者配置中删除先前的 partition.assignment.strategy 来 重新配置组中的每个消费者，仅保留 协作式 策略。
5. 依次重新启动组中的每个使用者，等待使用者在每次重启后重新加入组。

1. 将 Kafka Streams .jar 文件替换为新版本。
2. 在 Kafka Streams 配置中，将 upgrade.from 配置参数设置为您要从中升级的 Kafka 版本（例如 2.3）。
3. 依次重新启动每个流处理器（节点）。
4. 从 Kafka Streams 配置中删除 upgrade.from 配置参数。
5. 依次重新启动组中的每个消费者。