问Kafka Streams 2.5.0需要输入主题
EN

Stack Overflow用户

提问于 2020-04-21 11:31:29

回答 4查看 2.7K关注 0票数 6

从Kafka Streams 2.5.0开始，似乎拓扑必须包含一个输入主题。在Kafka 2.4.1 (及更早版本)中，情况并非如此。

我有一个应用程序，其中的拓扑只是创建一些全局状态存储，这些存储从其他应用程序写入的主题中读取数据。

在Kafka 2.5.0中，我得到了这个错误：

13:24:27.161 [<redacted>-7cf1b5c9-4a6e-4bf2-9f77-f7f85f2df3bb-StreamThread-1] ERROR o.a.k.s.p.internals.StreamThread - stream-thread [<redacted>-7cf1b5c9-4a6e-4bf2-9f77-f7f85f2df3bb-StreamThread-1] Encountered the following error during processing:
java.lang.IllegalStateException: Consumer is not subscribed to any topics or assigned any partitions
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1228)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1216)
    at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:853)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:753)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:697)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:670)

如果我添加一个虚拟输入主题(例如通过streamsBuilder.stream(Pattern.compile("hack"));)，应用程序启动正常。

这是意料之中的行为，还是Kafka Streams 2.5.0中的意外变化？

更多细节:上面的用例可能看起来有点奇怪，我不得不同意。这样做的原因首先是交互式查询的缺点，即应用程序在一段时间内无法回答查询。我看到这个问题已经通过KIP-535在Kafka Streans 2.5.0中得到了解决，这真是太棒了。我希望以后再研究一下IQ。

apache-kafka-streams

回答 4

Stack Overflow用户

回答已采纳

发布于 2020-05-11 18:45:33

当我们切换(回到)使用collection subscription时，在2.5.0中引入了一个回归。刚刚合并了一个fix，所以当它们发布时，你应该升级到2.5.1或2.6。

票数 3

Stack Overflow用户

发布于 2020-04-27 16:05:27

如果您的拓扑中没有任何非全局部分，那么就没有理由使用任何StreamThreads。这意味着您可以通过将num.threads设置为0来轻松解决此问题--可以说，您无论如何都应该这样做，以避免不必要的开销和组协调。当检测到全局拓扑时，默认情况下将其设置为零是一种“修复”方法，因此您不需要等待

票数 3

Stack Overflow用户

发布于 2020-04-22 11:56:16

private void subscribeConsumer() {
    if (builder.usesPatternSubscription()) {
        // this is old behaviour - is there a config that will revert to this??
        consumer.subscribe(builder.sourceTopicPattern(), rebalanceListener);
    } else {
        consumer.subscribe(builder.sourceTopicCollection(), rebalanceListener);
    }
}