每个人都希望他们的基础设施具有高可用性，ksqlDB也不例外。但是，像高可用性这样的关键特性，必须经过深思熟虑和严格的设计才能实现。我们认真考虑了如何在保证最少一次和精确一次处理的同时实现这一点。要了解我们是如何做到这一点的，现在是时候剥开ksqlDB下面的层，用Apache Kafka®再平衡协议来动手了。

消费者团体和再平衡协议

Kafka消费者组协议允许独立的资源管理和负载平衡，这是任何分布式系统对于应用程序开发人员和操作人员都必须具备的。通过指定一个代理(Broker)作为组的联络点，可以在组协调器中隔离所有组管理，并允许每个使用者只关注使用消息的应用程序级工作。

 

组协调器最终负责跟踪两件事:订阅主题的分区和组中的成员。对这些内容的任何更改都要求该组作出反应，以确保所有主题分区都是由其使用的，并且所有成员都是积极使用的。因此，当它检测到这些变化时，组织协调者就会拿出它唯一的工具:消费者组再平衡。

 

再平衡的前提简单而自我描述。所有成员被告知重新加入组，当前资源被刷新并“平均地”重新分配。当然，每个应用程序都是不同的，这意味着每个消费群体也是不同的。“均匀”平衡的负载可能意味着不同的事情，甚至对于共享相同代理的两个使用者组的组协调器也是如此。认识到这一点，再平衡协议早就被推到客户端，并完全从组协调器中抽象化了。这意味着不同的Kafka客户端可以插入不同的再平衡协议。

 

这篇博客文章主要关注消费者客户端和构建在消费者客户端的Kafka Streams。Konstantine Karantasis在他的博客文章《Apache Kafka中的增量合作再平衡:当你可以改变世界时，为什么要停止它》中详细讨论了Kafka Connect的协议。

 

为了让客户指定一组没有通信用户所遵循的协议，将选择一个成员作为重新平衡的组长，然后再平衡将分两个阶段进行。在第一阶段，组协调器等待每个成员加入组。这需要发送一个恰当命名的JoinGroup请求，其中每个成员编码一个订阅，包括任何感兴趣的主题和客户端定义的用户数据。订阅由代理合并，并在JoinGroup响应中发送给组长。

 

leader解码订阅，然后计算并编码分配给每个使用者的分区。然后在leader的SyncGroup请求中发送给组协调器。这是重新平衡的第二阶段:所有成员必须向代理发送一个SyncGroup请求，然后代理在SyncGroup响应中向它们发送分区分配。在整个再平衡阶段，成员之间从不直接交流。它们只能通过与经纪端组协调器对话来相互传播信息。

图1所示。有三个消费者的两阶段再平衡协议。在这种情况下，消费者C是组长。

 

使用者客户端进一步将给使用者的分区分配抽象为一个可插入的接口，称为ConsumerPartitionAssignor。如果分区转让者满足其消费者到所有分区的一对多映射契约，那么再平衡协议将负责其余的工作。这包括管理分区所有权从一个使用者到另一个使用者的转移，同时保证一个组中没有一个分区同时属于多个使用者。

 

这条规则听起来很简单，但在分布式系统中可能很难满足:“在同一时间”这个短语可能会在您的头脑中引起警报。

 

因此，为了使协议尽可能简单，迫切重新平衡协议诞生了:每个成员都需要在发送JoinGroup请求并参与重新平衡之前撤销其所有的分区。因此，该协议强制执行同步屏障;当JoinGroup响应被发送到leader时，所有的分区都是无主的，并且分区分配者可以自由地按照自己的意愿分配它们。

就安全性而言，这一切都是好的，但这个“停止世界”协议有严重的缺点:

• 在再平衡期间，该集团的任何成员都不能做任何工作
• 重新平衡持续时间随着分区计数的变化而变化，因为每个成员必须撤销并恢复分配的每个分区

每一个问题本身就已经够糟糕的了。综合起来，它们给消费者组协议的用户带来了一个严重的问题:在整个再平衡期间，所有分区和组的所有成员都将停机。这意味着从发送JoinGroup请求到接收SyncGroup响应，每个使用者都无所事事，如图2所示。

图2。迫切的再平衡协议。即使每个分区被重新分配给它的原始所有者，所有三个使用者都无法在虚线所指示的持续时间内使用分区。

 

虽然这在不经常重新平衡和资源很少的环境中是可以接受的，但大多数应用程序不属于这一类。我们已经做出了重大改进，减少了不稳定环境中不必要的再平衡。但只要有资源可管理，就需要重新平衡。也许我们可以通过更智能的客户端协议来减轻再平衡的痛苦。

 

理想的再平衡协议

要解决这个问题，请后退一步，问问在理想世界中最优的再平衡协议是什么样子的。设想一个没有节点故障、网络分区和竞争条件的世界——很好，对吗?现在考虑一下当组成员来来往往时重新分配分区的最简单方法。以这个特定的向外扩展的例子为例:

 

消费者A和B使用一个三分区主题。两个分区分配给A，一个分配给b。注意到负载不均匀，您决定添加第三个使用者。消费者C加入了这个群体。你希望这种再平衡如何发挥作用?

 

您所要做的就是将一个分区从消费者A移动到消费者c。所有三个消费者都必须重新加入组以确认它们的活动成员身份，但是对于消费者B来说，应该停止参与。在这个理想的世界中，使用者B没有理由停止处理它的分区，更不用说撤销和/或恢复它了。这意味着消费者B不会停机，如图3所示。

 

当然，消费者A必须撤销一个分区，但只有一个分区。一旦它放弃了它的第二个分区，它就可以愉快地返回来处理它的剩余分区。在整个重新平衡过程中，用户A并没有处于闲置状态，相反，它的停机时间只会持续到撤销一个分区所需的时间。

 

图3。你想要的再平衡协议。分区1和分区2被持续使用，分区3仅在将所有权从使用者A转移到C时停止工作。

 

总而言之，最佳再平衡只涉及那些需要移动以创建平衡分配的分区。您可以从最初的分配开始，并逐步移动分区以到达新的分配，而不是完全清除旧的分配以重新开始。

 

在理想世界和这里之间

不幸的是，回到现实世界的时候到了。消费者崩溃，不同步，拒绝合作。在这个美好的愿景完全消失之前，你所能希望的就是有一天能回来的具体计划。所以为什么不掸去旧地图绘制工具的灰尘，试着从你现在使用的渴望协议映射到一个更合作的协议呢?

 

要规划这一进程，你首先需要了解阻碍你前进的是什么。这又回到了上面介绍的规则:任何时候都不能有两个使用者声明同一分区的所有权。任何时候，只要放下同步屏障，就会面临风险。请记住，当前维护屏障的方法是强制所有成员在重新加入组之前撤销所有分区。

 

是不是太严格了?你大概可以猜到答案是肯定的。毕竟，这个障碍只需要对正在转移所有权的分区实施。重新分配给同一使用者的分区很容易满足该规则。其他分区则带来了挑战，因为使用者事先不知道他们的哪些分区将被重新分配到其他地方。显然，它们必须等待分区转让者确定分区到使用者的新映射。但是，一旦新的分配被发送到组的所有成员，同步就结束了。

 

如果要将一个分区从使用者a迁移到使用者B，那么B必须等待a放弃所有权后，B才能获得该分区。但是B无法知道A何时撤销了分区。请记住，吊销可以像从内存列表中删除分区那样简单，也可以像提交偏移量、刷新磁盘和清除所有相关资源那样复杂。例如，Kafka Streams用户启动状态严重的应用程序时非常清楚撤销一个分区需要多长时间。

 

显然，您需要某种方法让使用者指示何时重新分配其旧分区是安全的。但消费者只能在再平衡期间进行沟通，而最近一次再平衡刚刚结束。

 

当然，没有法律说你不能连续进行两次再平衡。如果你能让每次再平衡都不那么痛苦，那么第二次再平衡听起来就没那么糟糕了。这能以某种方式杠杆化到你想去的地方吗?渴望实现再平衡的隧道尽头是否有一线光明?

增量合作再平衡协议

好消息!您已经拥有了形成安全可靠的再平衡协议所需的所有组件。让我们退后一步，看看这种新的合作再平衡。

 

与前面一样，所有成员必须先发送一个JoinGroup请求。但这次，每个人都可以保留自己所有的分区。每个使用者并不撤销它们，而是在其订阅中对它们进行编码并将其发送给组协调器。

 

组协调器组装所有订阅并将它们发回给组组长，与前面一样。leader按照自己的意愿将分区分配给使用者，但是它必须从分配中删除所有正在转移所有权的分区。转让人可以利用每个订阅中编码的拥有分区来实施该规则。

 

从那里，新的分配—减去当前拥有的任何将要撤消的分区—传递给每个成员。使用者接受当前分配的差异，然后撤销新分配中没有出现的分区。同样地，他们将在分配中添加任何新的分区。对于出现在新旧分配中的每个分区，它们不需要做任何事情。很少有再平衡需要在用户之间进行分区迁移，因此在大多数情况下，几乎不需要做任何事情。

图4。分区的维恩图。既然可以撤销需要撤销的，为什么要撤销一切?

 

然后，被撤销分区的任何成员重新加入组，触发第二次再平衡，以便可以分配被撤销的分区。在此之前，这些分区是无主的和未分配的。同步障碍根本没有被消除;结果是，它只需要被移动。

 

在进行后续的再平衡时，所有已成功撤销的分区根据定义将不在已编码的拥有分区中。分割转让人可以自由地将它们分配给其合法的所有者。

图5。合作再平衡协议。通过进行第二次再平衡以强制同步障碍，可以避免撤销所有使用者的所有分区。

 

DIY合作再平衡

您可能已经猜到，这取决于分区转让人来完成这项工作。不太明显的是，这只取决于分区转让者——您可以通过简单地插入一个协作转让者来开启协作再平衡。

 

幸运的是，工具箱中已经添加了一个新的开箱即用分区转让者:CooperativeStickyAssignor。你可能对现有的粘性分派很熟悉——CooperativeStickyAssignor更进一步，它既粘性又具有协作性。

我们引入这个assignor是为了使Apache Kafka中的协作再平衡像设置配置一样简单，而不需要引入另一个客户端配置。但更微妙的动机在于把它包装成一个粘性转让者。通过这样做，我们可以保证让与人很好地遵守合作协议。

 

这意味着什么?为什么它如此重要?以RoundRobinAssignor为例，它不能很好地发挥合作再平衡。每次组成员关系或主题元数据更改时，轮询调度转让者生成的分配都会更改。它不试图粘贴返回分区到它们以前的所有者。但请记住，这是一个渐进的合作再平衡协议。整个算法通过一个分区一个分区地让分配增量地从旧的分配到新的分配来工作。如果新的赋值与之前的完全不同，那么增量变化就是整个赋值。你只会回到你开始的那个热切的协议，但是有更多的重新平衡。

 

所以粘性和合作在新转让人中同样重要。而且，由于订阅中编码了拥有的分区，粘贴就像以前一样容易。

 

除了具有粘性外，CooperativeStickyAssignor还必须确保从分配中删除所有必须撤销的分区。任何声称支持合作协议的转让人都必须履行该合同。所以，如果你想自己动手做一个定制的合作转让人，你可以从零开始写一个，甚至改编一个老的渴望转让人;只要确保它符合新规定就行了。

对于那些希望通过实时升级或转让者交换切换到新协议的人，最后要警告的是:遵循推荐的升级路径。滚动升级将触发重新平衡，你不希望陷入一半人遵循旧协议，另一半人遵循新协议的中间。有关如何安全升级到协作再平衡的更多细节，请参阅发布说明。

你能在Kafka流DIY吗?

如果你是一个Kafka流用户并且已经做到了这一步，你可能会想知道这一切是如何影响你的。当您不能选择使用哪个分区转让人时，您如何使用这个新协议?

 

幸运的是，你不必这么做。重新平衡协议的管理和选择嵌入到StreamsPartitionAssignor中，现在在默认情况下打开协作重新平衡。您只需要启动应用程序并观察它的运行。事实上，你甚至不需要去看它——但是监控你的应用程序始终是一个很好的练习。

 

那么这对流意味着什么呢?为了理解什么发生了改变，什么没有，让我们跟随一个虚构的卡夫卡流开发者的冒险——叫他弗朗茨。Franz正在运行一个使用Kafka流2.3版本的应用程序。他的拓扑使用了许多本地状态存储，他的输入主题有数百个分区，并且他的应用程序依赖于交互查询(IQ)特性在处理期间查询这些状态存储中的数据。

 

事情似乎进展顺利，但由于负责任的监控，Franz注意到他的实例正在以最大容量运行。是时候扩大规模了。不幸的是，添加新实例需要整个组重新平衡。Franz很沮丧地看到，每个实例已经停止处理每个分区，并且在重新平衡期间交互查询被禁用。此外，再平衡需要很长时间，考虑到需要关闭和重新打开许多州商店和分区，这一点也不奇怪。但是，他们真的需要吗?

 

幸运的是，他看到Kafka 2.4引入了一个新的再平衡协议，他希望这将有所帮助。Franz升级了他的Streams应用程序，小心地遵循发布说明中列出的特定升级路径，以确保安全滚动升级到合作协议。一旦群体稳定下来，它就试图再次扩大规模。

 

这一次，情况好多了。并不是每个商店都关闭并重新开业，只有在新实例上的少数分区被撤销。Franz还发现IQ在所有运行的实例上都是可用的，并且那些仍然在恢复数据到其状态存储的实例能够在整个再平衡过程中继续这样做。备用副本还会继续使用其存储的changelog主题，这使它们有时间赶上活动副本，并在发生故障时用很少的恢复时间接管。

 

更妙的是，当Franz升级到Kafka的最新版本时，他发现即使是正在运行的活动任务也可以在整个平衡过程中继续处理新记录，并且整个应用程序一直保持运行。他不再被迫在扩展和避免应用程序范围内停机之间做出选择。

 

渴望与合作:谁赢了?

尽管合作进行再平衡有明显的优势，但具体数字总是有最后发言权。因此，我们运行了一个基准来定量地比较两种再平衡协议并确定赢家。

 

我们用10个实例运行了一个简单的有状态Kafka流应用程序的基准测试。为了模拟一个常见的重新平衡触发场景，我们在组达到稳定运行状态后执行滚动反弹。应用程序在一段时间内的总吞吐量如下所示。这里的吞吐量是用每秒处理的记录数来度量的，通过误差条对10个实例进行汇总，以反映它们的方差。单从吞吐量的下降就可以很明显地看出滚动反弹的开始和结束。

图6。吞吐量(记录/秒)与时间(秒)对于经历滚动反弹的10个实例的迫切再平衡流应用程序。

图7。吞吐量(记录/秒)与时间(秒)对一个10个实例的协作再平衡应用程序进行滚动反弹。

 

绿色对应使用默认rocksdb支持的状态存储的应用程序，而红色对应内存中存储。由于重新平衡开销，协作RocksDB应用程序在弹跳期间的吞吐量只有轻微下降。有趣的是，协作内存中的应用程序似乎在吞吐量上仍然受到很大的影响，尽管它的恢复速度确实比急切的情况要快。这反映了在选择备份状态存储时的一种内在权衡:内存中的存储在稳定状态下速度更快，但在重启时遭受更大的挫折，因为它必须将所有临时数据从changelog主题恢复到内存中。

 

图6和图7中的图形为合作协议提供了一个视觉上引人注目的情况。但数据同样惊人:eager协议的暂停时间为37138毫秒，而cooperative协议的暂停时间仅为3522毫秒。当然，数字会因情况而异。我们鼓励您在您的应用程序中尝试合作再平衡，并自己衡量差异。

 

结论

从一开始，停止世界的再平衡协议就困扰着Kafka客户端的用户，包括Kafka Streams和堆栈上的ksqlDB。但是，重新平衡对于有效、均匀地分配资源是至关重要的，随着越来越多的应用程序迁移到云上并需要动态伸缩，这种情况只会越来越普遍。在渐进式的合作再平衡中，这未必会造成伤害。

 

基本的协作再平衡协议是在Apache Kafka 2.4版本中引入的。在版本2.5中添加了轮询新数据和释放增量协作再平衡的全部力量的能力。如果您一直生活在频繁重新平衡的阴影下，或者担心向外扩展会导致停机，那么请下载Confluent平台，它构建在Kafka的最新版本之上。

 

改善Kafka流和管理大量状态的ksqlDB应用程序可用性的进一步工作即将到来。使用KIP-441，在切换新实例之前，我们将开始在新实例上预热任务，从而缩小另一个随状态量而增大的可用性差距。与增量的协作再平衡相结合，它将允许您启动真正可伸缩和高可用性的应用程序——即使在再平衡期间也是如此。

 

无论您是从头开始构建一个普通的消费者应用程序，还是使用Kafka流进行一些复杂的流处理，还是使用ksqlDB解锁新的强大用例，消费者组协议都是应用程序的核心。这使得ksqlDB应用程序能够伸缩并平稳地处理故障。偷窥背后的真相是很有趣的，但是再平衡协议最好的部分是你不必这样做。作为用户，您所需要知道的是，您的ksqlDB应用程序可以在不牺牲高可用性的情况下容错。

原文：https://www.confluent.io/blog/cooperative-rebalancing-in-kafka-streams-consumer-ksqldb/

本文：http://jiagoushi.pro/node/1117

讨论：请加入知识星球【首席架构师圈】或者小号【jiagoushi_pro】