如何在Spring Cloud Stream Kafka Streams Binder中使用多路复用输入主题解决无效主题异常？

Question

我编写了一个Spring Cloud Streams Kafka Streams Binder应用程序，它将多个Kafka输入主题多路复用到一个流中:

spring:
  cloud:
    stream:
      bindings:
        process-in-0:
          destination: test.topic-a,test.topic-b

(来源:https://spring . io/blog/2019/12/03/stream-processing-with-spring-cloud-stream-and-Apache-Kafka-streams-part-2-programming-model-continued)

但是每当我在输入目的地(用逗号分隔)设置多个主题时，就会出现以下错误:

2022-06-17 14:07:07.648  INFO --- [-StreamThread-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=test-processor-2ba8d1d3-5bbe-45d3-a832-6a24cf2f5549-StreamThread-1-consumer, groupId=test-processor] Subscribed to topic(s): test-processor-KTABLE-AGGREGATE-STATE-STORE-0000000005-repartition, test.topic-a,test.topic-b  
2022-06-17 14:07:07.660  WARN --- [-StreamThread-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=test-processor-2ba8d1d3-5bbe-45d3-a832-6a24cf2f5549-StreamThread-1-consumer, groupId=test-processor] Error while fetching metadata with correlation id 2 : {test-processor-KTABLE-AGGREGATE-STATE-STORE-0000000005-repartition=UNKNOWN_TOPIC_OR_PARTITION, test.topic-a,test.topic-b=INVALID_TOPIC_EXCEPTION}  
2022-06-17 14:07:07.660 ERROR --- [-StreamThread-1] org.apache.kafka.clients.Metadata        : [Consumer clientId=test-processor-2ba8d1d3-5bbe-45d3-a832-6a24cf2f5549-StreamThread-1-consumer, groupId=test-processor] Metadata response reported invalid topics [test.topic-a,test.topic-b]  
2022-06-17 14:07:07.660  INFO --- [-StreamThread-1] org.apache.kafka.clients.Metadata        : [Consumer clientId=test-processor-2ba8d1d3-5bbe-45d3-a832-6a24cf2f5549-StreamThread-1-consumer, groupId=test-processor] Cluster ID: XYZ 
2022-06-17 14:07:07.663 ERROR --- [-StreamThread-1] org.apache.kafka.streams.KafkaStreams    : stream-client [test-processor-2ba8d1d3-5bbe-45d3-a832-6a24cf2f5549] Encountered the following exception during processing and Kafka Streams opted to SHUTDOWN_CLIENT. The streams client is going to shut down now.   

org.apache.kafka.streams.errors.StreamsException: org.apache.kafka.common.errors.InvalidTopicException: Invalid topics: [test.topic-a,test.topic-b]
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:627) ~[kafka-streams-3.2.0.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:551) ~[kafka-streams-3.2.0.jar:na]
Caused by: org.apache.kafka.common.errors.InvalidTopicException: Invalid topics: [test.topic-a,test.topic-b]

我尝试使用以下依赖项：

implementation 'org.apache.kafka:kafka-clients:3.2.0'
implementation 'org.apache.kafka:kafka-streams:3.2.0'
implementation "org.springframework.cloud:spring-cloud-stream"
implementation "org.springframework.cloud:spring-cloud-stream-binder-kafka"
implementation "org.springframework.cloud:spring-cloud-stream-binder-kafka-streams"
implementation "org.springframework.kafka:spring-kafka"

当我只设置一个输入主题时，一切正常。

我无法确定是什么导致了InvalidTopicException，因为我只在主题名称中使用了允许的字符，而且逗号分隔符似乎也是正确的(否则会出现不同的异常)。

Answer 1

实际上，在发布问题后，我自己找到了解决方案。因此，这里是未来的帮助：

显然，当我的处理器拓扑需要一个KTable作为输入类型时，我不能复用输入主题。当我将处理器签名更改为KStream时，它突然起作用：

不工作：

@Bean
public Function<KTable<String, Object>, KStream<String, Object>> process() {
  return stringObjectKTable ->
    stringObjectKTable
      .mapValues(...

工作：

@Bean
public Function<KStream<String, Object>, KStream<String, Object>> process() {
  return stringObjectKStream ->
    stringObjectKStream 
      .toTable()
      .mapValues(...

我不确定这是否是预期的行为，或者是否有其他问题，所以如果有更多的潜在问题，我希望得到任何提示。