如何替代/过渡一个Kafka消费群体？

Question

我有一个包含多个Kafka作品的资源库。我想将其中一个流提取到它自己的存储库中。但是，我不确定如何处理那个流的消费群体。我的意思是:在新的存储库中，流将有一个不同的< code>application.id。据我理解，消费者组的名称是基于< code>application.id设置的。如果我简单地关闭旧流，对于每个主题的每个分区，新流将从第零个偏移量开始，而不是从旧流停止的偏移量开始。这将导致输出主题中出现重复的消息。

是否有一些关于如何处理此问题的一般规则/最佳实践？我是否需要：

< li >关闭旧流； < li >检查每个主题的每个分区的旧流的使用者组偏移量； < li >“告诉”新流从各自的偏移量开始。我怎么“讲”:)它？我的意思是——如果我使用< code > Kafka-console-consumer ，有< code>partition和< code>offset选项，但是流在后台启动自己的consumer，所以我不确定如何控制它。

流使用了相当多的输入主题（大约20个，幸运的是每个主题都有一个分区），因此我不确定如何准确地处理这个问题。

Answer 1

（0）如果可能，我建议尝试保留application.id，所有问题都会消失。在这种情况下，您可以先停止旧应用程序，然后启动新应用程序，甚至可以先启动所有新实例，然后拆除所有旧实例。由于新旧应用程序都连接到同一个Kafka集群，因此将进行无缝切换。

（1） 只有当Kafka Streams应用程序是无状态的时，您想做的事情才有效；对于有状态的应用程序，需要更多的步骤来将状态传递到新的应用程序实例。或者，新实例将以空状态开始（根据应用程序的要求，这也可以）。

(2)首先，您需要停止所有旧的应用程序实例，并接收旧的< code>application.id的每个输入主题分区的最后提交的偏移量(通过< code > bin/Kafka-consumer-groups . sh 使用< code> - describe - group选项)，然后，您需要使用新的< code>application.id提交这些偏移量(同样，您可以使用< code > bin/Kafka-consumer-groups . sh 使用< code> - to-offset选项)。(有关https://cwiki.apache.org/confluence/display/KAFKA/KIP-122:添加重置用户组偏移工具的详细信息)之后，您可以启动新的应用程序实例，该实例将为新的< code>application.id获取提交的偏移。