Kafka迭代升级，轮询升级

怼烎@ 2022-12-25 04:52 204阅读 0赞

**滚动升级过程中特别容易忽略的一点：broker之间的通讯协议和message的传输协议要与旧版本的一致，否则升级完的broker会因为通讯协议版本不一致导致节点一直报错（**Connection to "broker id" was disconnected before the response was read**）**

# 通讯协议版本
    inter.broker.protocol.version=2.6-IV0
    # 消息格式版本
    log.message.format.version=2.6-IV0

如下是官方给出的可识别值

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70][]

# 2.6升级步骤 #

如果您正在从2.1.x 之前的版本升级。请参阅下面关于用于存储消费者偏移量的模式更改的注释。一旦您更改了 inter.broker.protocol.version 到最新版本时，将不可能降级到2.1之前的版本。

## 滚动升级 ##

1、更新 server.properties，并添加以下属性。CURRENT\_KAFKA\_VERSION 指的是您正在升级的版本。CURRENT\_MESSAGE\_FORMAT\_VERSION 指的是当前使用的消息格式版本。如果您之前覆盖了消息格式版本，则应该保持其当前值。或者，如果您正在从0.11.0.x 之前的版本升级。那么CURRENT\_MESSAGE\_FORMAT\_VERSION应该设置为匹配CURRENT\_KAFKA\_VERSION。

*  inter.broker.protocol.version=CURRENT\_KAFKA\_VERSION(例如，2.5,2.4等)
 *  log.message.format.version=CURRENT\_MESSAGE\_FORMAT\_VERSION(有关此配置的详细信息，请参阅升级后对性能的潜在影响)。

如果您正在从版本0.11.0.x 升级或以上版本，并且您没有覆盖消息格式，那么您只需要覆盖代理间协议版本。

*  inter.broker.protocol.version=CURRENT\_KAFKA\_VERSION(例如，2.5,2.4等)

2、一次升级一个 Brokers：关闭代理，更新代码，并重新启动它。完成之后，代理将运行最新版本，您可以验证集群的行为和性能是否满足预期。如果有任何问题，在这一点上仍然有可能降级。

3、在验证了集群的行为和性能之后，通过编辑 inter.broker.protocol.version 来修改协议版本。并将其设置为2.6。

4、一个接一个地重新启动代理，使新协议版本生效。一旦代理开始使用最新的协议版本，就不可能将集群降级为旧版本。

5、如果您按照上面的指示覆盖了消息格式版本，那么您需要再进行一次滚动重启，以将其升级到最新版本。一旦所有(或大多数)使用者都升级到0.11.0或更高版本，请更改 log.message.format.version 升级到2.6，并逐个重新启动它们。请注意，旧的Scala客户机(不再维护)不支持0.11中引入的消息格式，因此为了避免转换成本(或利用准确的一次语义)，必须使用新的Java客户机。

## 2.6.0中值得注意的变化 ##

*  Kafka Streams添加了一种新的处理模式(要求broker 2.5或更新)，这种模式通过一次保证提高了应用程序的可伸缩性(cf. [KIP-447][])
 *  对于Java 11或更新版本，TLSv1.3是默认启用的。客户机和服务器将协商是否支持TLSv1.3，否则将退回到TLSv1.2。See [KIP-573][] for more details.
 *  client.dns.lookup 的默认值已经从 default 改成 use\_all\_dns\_ips，如果主机名解析为多个IP地址，客户端和代理现在将依次尝试连接到每个IP，直到连接成功建立。See [KIP-602][] for more details.
 *  `NotLeaderForPartitionException` 弃用并替换为 `NotLeaderOrFollowerException。`Fetch 请求和其他仅针对 leader 或 follower 的请求返回 NOT\_LEADER\_OR\_FOLLOWER(6)，而不是 REPLICA\_NOT\_AVAILABLE(9)，如果代理不是副本，确保重新分配期间的瞬时错误由所有客户端作为可检索异常处理。

# 2.5升级步骤 #

从[0.8.x][], 0.9.x, 0.10.0.x, 0.10.1.x, 0.10.2.x, 0.11.0.x, 1.0.x, 1.1.x, 2.0.x or 2.1.x or 2.2.x, 2.3.x, 2.4.x 升级到 2.5.0

如果您正在从2.1.x之前的版本升级。请参阅下面关于用于存储消费者偏移量的模式更改的注释。一旦您更改了 inter.broker.protocol.version 到最新版本时，将不可能降级到2.1之前的版本。

## 滚动升级 ##

1、更新 server.properties，并添加以下属性。CURRENT\_KAFKA\_VERSION 指的是您正在升级的版本。CURRENT\_MESSAGE\_FORMAT\_VERSION 指的是当前使用的消息格式版本。如果您之前覆盖了消息格式版本，则应该保持其当前值。或者，如果您正在从0.11.0.x 之前的版本升级。那么 CURRENT\_MESSAGE\_FORMAT\_VERSION 应该设置为匹配 CURRENT\_KAFKA\_VERSION。

*  inter.broker.protocol.version=CURRENT\_KAFKA\_VERSION (e.g. 0.10.0, 0.11.0, 1.0, 2.0, 2.2).
 *  log.message.format.version=CURRENT\_MESSAGE\_FORMAT\_VERSION

如果您正在从版本0.11.0.x 或以上版本升级，并且您没有覆盖消息格式，那么您只需要覆盖代理间协议版本。

*  inter.broker.protocol.version=CURRENT\_KAFKA\_VERSION (0.11.0, 1.0, 1.1, 2.0, 2.1, 2.2, 2.3).

2、一次升级一个 brokers:关闭代理，更新代码，并重新启动它。完成之后，代理将运行最新版本，您可以验证集群的行为和性能是否满足预期。如果有任何问题，在这一点上仍然有可能降级。

3、在验证了集群的行为和性能之后，通过编辑 inter.broker.protocol 来修改协议版本，并设置为2.5。

4、一个接一个地重新启动 Brokers，使新协议版本生效。一旦代理开始使用最新的协议版本，就不可能将集群降级为旧版本。

5、如果您按照上面的指示覆盖了消息格式版本，那么您需要再进行一次滚动重启，以将其升级到最新版本。一旦所有(或大多数)使用者都升级到 0.11.0 或更高版本，请更改 log.message.format.version 升级到2.5，并逐个重新启动它们。请注意，旧的Scala客户机(不再维护)不支持0.11中引入的消息格式，因此为了避免转换成本(或利用准确的一次语义)，必须使用新的Java客户机。

## 2.5.0中值得注意的变化 ##

*  当使用 RebalanceProtocol\#COOPERATIVE 时，Consumer\#poll 仍然可以返回数据，尽管它还在重新平衡那些仍由消费者拥有的分区的过程中；此外，Consumer\#commitSync 现在可能会抛出一个非致命的 RebalanceInProgressException 来通知用户这样的事件，以便与致命的 CommitFailedException 区分开来，并允许用户完成正在进行的再平衡，然后重新尝试提交那些仍然拥有的分区的偏移量。
 *  为了提高典型网络环境中的弹性，请将 zookeper.session.timeout.ms 的默认值从6s改成18s，replica.lag.time.max.ms 从10改为30s。
 *  添加了新的 DSL 算子 cogroup()，用于一次性将多个流聚合在一起。
 *  添加了新的 KStream.toTable() API，用于将输入事件流转换为 KTable。
 *  添加了一个新的 Serde 类型 Void 来表示来自输入主题的空键或空值。
 *  废弃了 `UsePreviousTimeOnInvalidTimestamp` 并替换为 `UsePartitionTimeOnInvalidTimeStamp`.
 *  通过添加挂起的偏移隔离机制和更强的事务提交一致性检查，改进了精确一次语义，这极大地简化了可伸缩的精确一次应用程序的实现。我们还在样例文件夹下添加了一个新的仅一次语义代码示例。 Check out [KIP-447][] for the full details.
 *  添加了一个新的公共api KafkaStreams.queryMetadataForKey(String, K, Serializer) 来获取被查询的键的详细信息。除了包含密钥的活动分区和备用分区的主机外，它还提供了关于密钥所在分区的信息。
 *  `通过弃用 KafkaStreams.store(String, QueryableStoreType)` 并替换为  `KafkaStreams.store(StoreQueryParameters)，`支持查询过时的存储(以获得高可用性)和属于特定分区的存储。
 *  通过 kafkastreams.alllocalstorepartitionlag() 添加了一个新的公共 api 来访问本地存储到一个实例的延迟信息。
 *  不再支持Scala 2.11。See [KIP-531][] for details.
 *  kafka.security.auth 包中所有 Scala 类被弃用，有关在2.4.0中添加的新的Java授权器API的详细信息See [KIP-504][] ，`kafka.security.auth.Authorizer` 和 `kafka.security.auth.SimpleAclAuthorizer` 再 2.4.0 已经被弃用。
 *  TLSv1和TLSv1.1在默认情况下是禁用的，因为它们存在已知的安全漏洞。现在默认情况下只有TLSv1.2是启用的。您可以继续使用TLSv1和TLSv1.1，方法是在配置选项ssl中显式地启用它们 `ssl.protocol` 和 `ssl.enabled.protocols。`
 *  ZooKeeper 升级到 3.5.7，如果3.4.x 数据目录中没有快照文件，ZooKeeper从3.4.x升级到3.5.7可能会失败，这通常发生在测试升级中，当ZooKeeper 3.5.7尝试加载一个没有创建快照文件的已有3.4 data dir时。For more details about the issue please refer to [ZOOKEEPER-3056][]. A fix is given in [ZOOKEEPER-3056][], which is to set `snapshot.trust.empty=true` config in `zookeeper.properties` before the upgrade。一个临时的解决方案：升级前将 `snapshot.trust.empty=true在zookeeper.properties。`
 *  ZooKeeper版本3.5.7支持tls加密连接到ZooKeeper，无论是否有客户端证书，还有额外的Kafka配置可以利用这一点。See [KIP-515][] for details.

更多：[http://kafka.apache.org/documentation/\#upgrade][http_kafka.apache.org_documentation_upgrade]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70]: /images/20221120/47ef4adeec3347e99ce77886602a7282.png
[KIP-447]: https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+Producer+scalability+for+exactly+once+semantics
[KIP-573]: https://cwiki.apache.org/confluence/display/KAFKA/KIP-573%3A+Enable+TLSv1.3+by+default
[KIP-602]: https://cwiki.apache.org/confluence/display/KAFKA/KIP-602%3A+Change+default+value+for+client.dns.lookup
[0.8.x]: http://kafka.apache.org/25/documentation.html#upgrade_2_5_0
[KIP-531]: https://cwiki.apache.org/confluence/display/KAFKA/KIP-531%3A+Drop+support+for+Scala+2.11+in+Kafka+2.5
[KIP-504]: https://cwiki.apache.org/confluence/display/KAFKA/KIP-504+-+Add+new+Java+Authorizer+Interface
[ZOOKEEPER-3056]: https://issues.apache.org/jira/browse/ZOOKEEPER-3056
[KIP-515]: https://cwiki.apache.org/confluence/display/KAFKA/KIP-515%3A+Enable+ZK+client+to+use+the+new+TLS+supported+authentication
[http_kafka.apache.org_documentation_upgrade]: http://kafka.apache.org/documentation/#upgrade