Kubernetes 上的 Kafka Streams：重新部署后的长期重新平衡

Question

问题

我们使用 StatefulSet 在 Kubernetes 上部署 Scala Kafka Streams 应用程序。这些实例具有单独的applicationIds，因此它们每个都复制完整的输入主题以实现容错。它们本质上是只读服务，仅读取状态主题并将其写入状态存储，从该存储中通过 REST 服务客户请求。这意味着，在任何给定时间，消费者组始终只包含一个单个 Kafka Streams 实例。

我们现在的问题是，在触发滚动重启时，每个实例大约需要 5 分钟才能启动，其中大部分时间都花在等待 REBALANCING 状态。我读过here，Kafka Streams 不会发送LeaveGroup 请求，以便在容器重启后快速返回，无需重新平衡。为什么这对我们不起作用？为什么重新平衡需要这么长时间，即使 applicationId 是相同的？理想情况下，为了最大限度地减少停机时间，应用程序应立即从其重新启动时离开的位置接管。

配置

以下是我们从默认值更改的一些配置：

properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), "1000")
properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG), "300000")
properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), "earliest")
// RocksDB config, see https://docs.confluent.io/current/streams/developer-guide/memory-mgmt.html
properties.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, classOf[BoundedMemoryRocksDBConfig])    

问题/相关配置

• 减少session.timeout.ms 会有所帮助吗？我们将其设置为相当大的值，因为 Kafka 代理位于不同的数据中心，并且网络连接有时不是超级可靠。
• This answer 建议减少 max.poll.interval.ms，因为它与重新平衡超时有关。那是对的吗？我很犹豫是否要更改此设置，因为它可能会对我们应用的正常运行模式产生影响。
• There is mention 配置 group.initial.rebalance.delay.ms 在部署期间延迟重新平衡 - 但这也会在从崩溃中恢复后导致延迟，不是吗？
• 我还偶然发现了KIP-345，它旨在完全通过group.instance.id 消除消费者对静态会员的重新平衡，这非常适合我们的用户案例，但它似乎还没有在我们的代理上可用。

我对大量配置以及如何使用它们在更新后启用快速恢复感到困惑。谁能解释一下他们是怎么一起玩的？

Answer 1

您引用的另一个问题并不是说在重新启动时避免重新平衡。不发送LeaveGroupRequest 只会在您停止应用程序时避免重新平衡。因此，重新平衡的数量从两个减少到一个。当然，对于你有点不寻常的单实例部署，你在这里没有任何收获（事实上，它实际上可能会“伤害”你......）a

减少 session.timeout.ms 有帮助吗？我们将其设置为相当大的值，因为 Kafka 代理位于不同的数据中心，并且网络连接有时不是超级可靠。

可能，取决于您重启应用的速度。 （下面有更多详细信息。）也许只是尝试一下（即，将其设置为 3 分钟以仍然具有较高的稳定性值，然后看到重新平衡时间下降到 3 分钟？

这个答案建议减少 max.poll.interval.ms，因为它与重新平衡超时有关。那是对的吗？我很犹豫是否要更改此设置，因为它可能会对我们应用的正常运行模式产生影响。

max.poll.interval.ms 也会影响重新平衡时间（下面有更多详细信息）。但是，默认值为 30 秒，因此不应导致 5 分钟的重新平衡时间。

提到了一个配置 group.initial.rebalance.delay.ms 在部署期间延迟重新平衡 - 但这也会在从崩溃中恢复后导致延迟，不是吗？

这仅适用于空消费者组，默认值仅为 3 秒。所以应该不会影响到你。

我还偶然发现了 KIP-345，它旨在完全通过 group.instance.id 消除消费者对静态成员的重新平衡，这非常适合我们的用户案例，但它似乎还没有在我们的经纪人。

使用静态组成员身份实际上可能是最好的选择。也许值得升级您的经纪人以获得此功能。

顺便说一句，session.timeout.ms 和 max.poll.interval.ms 之间的区别在另一个问题中解释：Difference between session.timeout.ms and max.poll.interval.ms for Kafka 0.10.0.0 and later versions

通常，代理端组协调器维护每个“组生成”的所有成员的列表。如果成员主动离开群组（通过发送LeaveGroupRequest）、超时（通过session.timeout.ms 或max.poll.interval.ms）或新成员加入群组，则会触发重新平衡。如果发生重新平衡，每个成员都有机会重新加入该组以包含在下一代中。

对于您的情况，该组只有一名成员。当你停止你的应用程序时，不会发送LeaveGroupRequest，因此只有在session.timeout.ms 传递之后，组协调器才会删除该成员。

如果您重新启动应用程序，它会作为“新”成员返回（从组协调员的角度来看）。这将触发重新平衡，使该组的所有成员都可以更改以重新加入该组。对于您的情况，“旧”实例可能仍在组中，因此重新平衡只会在组协调器从组中删除旧成员后继续进行。问题可能是，组协调员认为组从一个成员扩展到两个成员......（这就是我上面的意思：如果发送LeaveGroupRequest，当你阻止你时，组将变为空应用程序，并且在重新启动时，只有新成员会在组中，并且重新平衡将立即进行。）

使用静态组成员身份可以避免该问题，因为在重新启动时实例可能会被重新识别为“旧”实例，并且组协调器不需要等待旧组成员过期。