Kafka - 冗余应用程序的拓扑更改

Question

假设我有两个应用程序具有相同的 applicationId“foo-processor”和以下设置：

    streamsBuilder.table(fooTopic)
                  .groupBy(...)
                  .reduce(...)

假设我现在有一些我不想处理的情况并添加这样的过滤器：

        streamsBuilder.table(fooTopic)
                      .filter(...)
                      .groupBy(...)
                      .reduce(...)

在部署期间，并非应用的所有实例都会同时关闭和重新启动。因此，foo-processor 的实例#1 重新启动，而实例#2 仍使用先前的拓扑。发生的情况是实例 #1 将出现此错误：

java.lang.IllegalArgumentException: Assigned partition foo-processor-KTABLE-REDUCE-STATE-STORE-0000000006-repartition-2 for non-subscribed topic regex pattern; subscription pattern is foo-processor-KTABLE-REDUCE-STATE-STORE-0000000007-repartition|<topic>

我认为这是预期的行为，因为由于拓扑不同，重新分区主题可能不包含相同的事件。话虽这么说，我想知道我应该如何处理拓扑的变化。

• 这是否意味着应用程序不同，因此 applicationId 也应该更改？如果没有，如果同一应用的多个实例正在运行，我应该如何处理拓扑更改？

谢谢！

Answer 1

如果要更改拓扑，则需要使用新的application.id -- 不支持与相同的application.id 并行运行。