在具有多个主题分区的 Kafka Streams 中无法重新平衡错误

Question

当源主题分区计数 = 1 时工作正常。如果我将分区增加到任何大于 1 的值，我会看到以下错误。适用于 Low level 以及 DSL API。任何指针？可能缺少什么？

org.apache.kafka.streams.errors.StreamsException: stream-thread [StreamThread-1] Failed to rebalance
        at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:410)
        at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:242)

Caused by: org.apache.kafka.streams.errors.StreamsException: task [0_1] Store in-memory-avg-store's change log (cpu-streamz-in-memory-avg-store-changelog) does not contain partition 1
        at org.apache.kafka.streams.processor.internals.ProcessorStateManager.register(ProcessorStateManager.java:185)
        at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.register(ProcessorContextImpl.java:123)
        at org.apache.kafka.streams.state.internals.InMemoryKeyValueStoreSupplier$MemoryStore.init(InMemoryKeyValueStoreSupplier.java:102)
        at org.apache.kafka.streams.state.internals.InMemoryKeyValueLoggedStore.init(InMemoryKeyValueLoggedStore.java:56)
        at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.init(MeteredKeyValueStore.java:85)
        at org.apache.kafka.streams.processor.internals.AbstractTask.initializeStateStores(AbstractTask.java:81)
        at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:119)
        at org.apache.kafka.streams.processor.internals.StreamThread.createStreamTask(StreamThread.java:633)
        at org.apache.kafka.streams.processor.internals.StreamThread.addStreamTasks(StreamThread.java:660)
        at org.apache.kafka.streams.processor.internals.StreamThread.access$100(StreamThread.java:69)
        at org.apache.kafka.streams.processor.internals.StreamThread$1.onPartitionsAssigned(StreamThread.java:124)
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:228)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:313)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:277)
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:259)
        at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1013)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:979)
        at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:407)

Answer 1

这是一个操作问题。 Kafka Streams 不允许在其“生命周期”期间更改输入主题分区的数量。

如果您停止正在运行的 Kafka Streams 应用程序，更改输入主题分区的数量，然后重新启动您的应用程序，它将中断（您会在上面看到错误）。在生产用例中解决这个问题很棘手，强烈建议不要更改输入主题分区的数量（参见下面的评论）。不过，对于 POC/演示，修复起来并不难。

为了解决这个问题，您应该使用 Kafka 的应用程序重置工具重置您的应用程序：

• http://docs.confluent.io/current/streams/developer-guide.html#application-reset-tool
• https://www.confluent.io/blog/data-reprocessing-with-kafka-streams-resetting-a-streams-application/

使用应用程序重置工具的缺点是会清除整个应用程序状态。因此，为了使您的应用程序进入与以前相同的状态，您需要从头开始重新处理整个输入主题。这当然只有在所有输入数据仍然可用并且应用主题保留时间/大小策略的代理没有删除任何内容的情况下才有可能。

此外，您应该注意，将分区添加到输入主题会更改主题的分区模式（默认基于哈希键的分区）。因为 Kafka Streams 假设输入主题按键正确分区，所以如果您使用重置工具并重新处理所有数据，您可能会得到错误的结果，因为“旧”数据与“新”数据的分区方式不同（即，添加后写入的数据新分区）。对于生产用例，您需要从原始主题中读取所有数据并将其写入新主题（分区数量增加）以正确分区您的数据（或者当然，此步骤可能会更改不同记录的顺序键——通常不应该是一个问题——只是想提一下）。之后，您可以将新主题用作 Streams 应用的输入主题。

这个重新分区步骤也可以在您的 Streams 应用程序中轻松完成，方法是在阅读原始主题之后和进行任何实际处理之前直接使用运算符 through("new_topic_with_more_partitions")。

但是，一般来说，建议您对生产用例的主题进行过度分区，这样您以后就无需更改分区数量。过度分区的开销相当小，可以为您以后省去很多麻烦。如果您使用 Kafka，这是一个一般性建议——它不仅限于 Streams 用例。

再说一句：

有些人可能会建议手动增加Kafka Streams内部主题的分区数。首先，这是一种hack，出于某些原因不推荐。

1. 可能很难确定正确的数字是多少，因为它取决于各种因素（因为它是 Stream 的内部实现细节）。
2. 您还面临破坏分区方案的问题，如上段所述。因此，您的应用程序很可能以不一致的状态结束。

为了避免应用状态不一致，Streams 不会自动删除任何内部主题或更改内部主题的分区数，但会失败并显示您报告的错误消息。这样可以确保用户通过手动执行“清理”来了解所有影响。

顺便说一句：对于即将推出的 Kafka 0.10.2，此错误消息得到了改进：https://github.com/apache/kafka/blob/0.10.2/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java#L100-L103