如何在部署期间在Kafka Streams应用程序中实现高可用性？

Question

主要问题：我们在库伯内特斯上运行Kafka Streams（Java）应用程序，以在我们的Kafka集群（运行Confluent Community Edition v7.0/Kafka v3.0）中消耗、处理和生成实时数据。我们如何以限制消耗记录停机时间的方式部署我们的应用程序？我们最初的目标是每个任务停机时间约为2秒。

我们的目标是持续部署生产环境的更改，但部署的破坏性太大，导致应用程序中的记录消耗停机，导致生成的实时记录延迟。

我们尝试了不同的策略来了解它如何影响延迟（停机时间）。

策略#1：

< li >终止所有应用程序实例(共6个) < li >立即启动所有新的应用程序实例 < li >结果:已使用记录的测量最大延迟:< code>85秒

策略 #2：

< li >启动一个新的应用程序实例 < li >等待< code>3分钟以允许在新的应用程序实例中恢复本地状态 < li >在< code>3分钟后终止一个旧的应用程序实例 < li >重复此操作，直到所有旧的应用程序实例都终止为止 < li >结果:测量的已使用记录的最大延迟:< code>39秒

策略 #3：

• 与策略#2相同，但将等待时间增加到15分钟
• 结果：测量消耗记录的最大延迟：7秒。但是每个应用程序实例15分钟将导致15分钟x 6个实例=90分钟部署更改额外的30分钟以完成增量重新平衡协议。我们发现部署时间相当长。

我们一直在阅读KIP-429：Kafka消费者增量再平衡协议，并试图配置该应用程序以支持我们的用例。

以下是我们为策略#2和#3所做的关键Kafka Streams配置：

acceptable.recovery.lag: 6000
num.standby.replicas: 2
max.warmup.replicas: 6
probing.rebalance.interval.ms: 60000
num.stream.threads: 2

输入主题有< code>12个分区，平均消息速率为< code>800条记录/秒。有3个Kafka Streams键值状态存储，其中两个具有与输入主题相同的速率。这两个大约是< code>20GB大小。密钥集的大小约为< code>4000。理论上，上述< code > acceptable . recovery . lag 在每个分区的changelog主题上应该有< code>~60秒的延迟。

我们做了一些值得注意的观察:

1a-第一个新应用实例启动

1b-Kafka立即重新平衡，2个旧的应用程序实例被分配了更多的任务，2个旧的任务丢失了

1c-同时记录的最大延迟从0.2秒增加到3.5秒（这表明重新平衡大约需要3秒）

2-探测重新平衡发生，Kafka Streams以某种方式决定从旧应用程序实例之一撤销任务并将其交给已经拥有最多任务的实例

< code>3 -最后一个旧应用程序实例被终止

4 - 所有分区都像升级前一样重新平衡，增量重新平衡完成（在最后一个应用程序实例终止后约 33 分钟）

其他-为第一个新应用程序实例分配任务大约需要40分钟。此外，每个任务都被多次重新分配，导致许多小的中断3秒。

如果需要，我们可以提供更多关于其他策略的拓扑、主题、配置和度量图的详细信息（这个线程已经非常庞大）。




Answer 1

给你的Kafka Stream应用的一些建议(不是专家，但我个人观察到的)。

1. 将pod从部署更改为StatefulSet并添加此配置group.instance.id："${hostname}"。通过这种方式，pod将保持pod的相同名称，您将能够使用Kafka消费者增量再平衡协议
2. 当您现在使用StatefulSet时，请将状态存储保存在永久存储中，它会从Kafka中删除重新加载的状态存储。但是请注意，您需要根据库伯内特斯接收到的信号正确关闭Kafka流，并调整终止GracePerod秒以允许正确关闭。如果没有，Kafka Stream将检测到检查点不干净，并将重新获取状态存储（不清楚您的状态存储是否已保存，但我认为已经完成了）。

我所观察到的唯一一件事是，简单的消费群体再平衡比你预期的2秒钟长，这似乎是一个复杂的目标（就像你所说的再平衡需要3秒钟）。但是使用增量重新平衡，我认为在重新平衡过程中会处理一些分区（目前还没有亲自测量）。

Answer 2

我无法发表评论（因为对SO相对较新），所以我将在这里回复。您关于状态集和滚动重启的观点是正确的，但您可以通过使用 podManagementPolicy：在有状态集中并行来解决此问题。这使得它与部署相同，因为所有 Pod 将同时出现。

https://kubernetes.io/docs/concepts/workloads/controllers/statefulset/#pod-management-policies