连接到Kafka的Spring Cloud Stream的多个StreamListeners

Question

在一个使用Spring Cloud Stream连接到Kafka的Spring Boot应用程序中，我试图设置两个单独的Stream侦听器方法：

null

Exception in thread "test-d44cb424-7575-4f5f-b148-afad034c93f4-StreamThread-2" java.lang.IllegalArgumentException: Assigned partition t1-0 for non-subscribed topic regex pattern; subscription pattern is t3
    at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignFromSubscribed(SubscriptionState.java:195)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:225)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:367)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295)
    at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1146)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1111)
    at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:848)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:805)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:771)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:741)

我认为重要的部分是：“为非订阅的主题正则表达式模式分配分区t1-0；订阅模式是T3”。这是两个不相关的主题，所以据我所知，任何与t3相关的东西都不应该订阅任何与T1相关的东西。引起问题的确切主题也会断断续续地发生变化：有时被提及的是自动生成的主题之一，而不是t1本身。

以下是两个流侦听器的设置方式（以Kotlin表示）：

@StreamListener
fun listenerForT1AndT2(
        @Input("t1") t1KTable: KTable<String, T1Obj>,
        @Input("t2") t2KTable: KTable<String, T2Obj>) {

    t2KTable
        .groupBy(...)
        .aggregate(
                { ... },
                { ... },
                { ... },
                Materialized.with(Serdes.String(), JsonSerde(SomeObj::class.java)))
        .join(t1KTable,
                { ... },
                Materialized.`as`<String, SomeObj, KeyValueStore<Bytes, ByteArray>>("test")
                        .withKeySerde(Serdes.String())
                        .withValueSerde(JsonSerde(SomeObj::class.java)))
}

@StreamListener
fun listenerForT3(@Input("t3") t3KStream: KStream<String, T3Obj>) {
    events.map { ... }
}

但是，当我只使用@streamlistener注释一个方法来设置代码，并为所有三个主题取参数时，一切工作都很好，例如。

@StreamListener
fun compositeListener(
        @Input("t1") t1KTable: KTable<String, T1Obj>,
        @Input("t2") t2KTable: KTable<String, T2Obj>,
        @Input("t3") t3KStream: KStream<String, T3Obj>) {
    ...
}

我还尝试将这两个侦听器方法分成两个单独的类，每个类只对它感兴趣的接口使用@enablebinding（即t1和t2的一个接口，t3的一个单独的接口），但这没有帮助。

我发现的所有与这个错误消息相关的东西，例如这里，都是关于拥有多个应用程序实例的，但在我的例子中，只有一个Spring Boot应用程序实例。

Answer 1

每个StreamListener方法都需要单独的应用程序id。这里有一个例子：

spring.cloud.stream.kafka.streams.bindings.t1.consumer.application-id=processor1-application-id spring.cloud.stream.kafka.streams.bindings.t2.consumer.application-id=processor1-application-id spring.cloud.stream.kafka.streams.bindings.t3.consumer.application-id=processor2-application-id

您可能希望使用最新的快照(2.1.0)进行测试，因为绑定器处理应用程序id的方式最近发生了一些变化。

更多详情请看这里的更新。下面是多个StreamListener方法的工作示例，这些方法是Kafka流处理器。