多主题流监听kafka之间通信出错|未从 .properties 文件中获取应用程序 ID

Question

我在与 kafka 流主题通信时遇到错误。当我尝试从第一个主题到第二个主题时，它工作正常。但是在从第二个主题发送到第三个主题时，它会出错，说 Failed to create consumer binding , retry in 30 seconds , Null pointer exception at registerconsumerproperties 在调试时我意识到，第三个主题的 applicationID 被传递为 null ，尽管我已经明确指定了它

我的application.properties文件

## Topic 1
-  spring.cloud.stream.bindings.input1.destination=topic-1
-  spring.cloud.stream.bindings.input1.consumer.applicationId=processor-1


## Topic 2
-  spring.cloud.stream.bindings.output2.bindings=<topic-2 name>
-  spring.cloud.stream.bindings.input2.destination=<topic-2 name >
-  spring.cloud.stream.bindings.input2.consumer.applicationId=processor-2

## Topic 3
-  spring.cloud.stream.bindings.output3.destination=<topic-3 name>
-  spring.cloud.stream.bindings.input3.destination=<topic-3 name>
-  spring.cloud.stream.bindings.input3.consumer.applicationId=processor-3

1st listener

public classA {
@StreamListener
public void process(@INPUT(check.Const_A) KStream<String, StreamClass> stream)
 stream.foreach((key,value)-> System.out.println(key + value)));
stream.to(check.Const_B);
}

2nd listener
public classB {
@StreamListener
public void process(@INPUT(check.Const_C) KStream<String, StreamClass> stream)
 stream.foreach((key,value)-> System.out.println(key + value)));
stream.to(check.Const_D);
}

3rd listener
public classC {
@StreamListener
public void process(@INPUT(check.Const_E) KStream<String, StreamClass> stream)
 stream.foreach((key,value)-> System.out.println(key + value)));

}

Interface (to be added in @EnableBinding annotation)

public check {
String Const_A = "input1"
String Const_B="output2"
String CONST_C="input2"
String CONST_D="output3"
String CONST_E="input3"

@Input(Const_A)
KStream<String,StreamClass> inATopic();

@Output(Const_B)
MessageChannel outATopic();

@Input(Const_C)
KStream<String,StreamClass> inBTopic();

@Output(Const_D)
MessageChannel outBTopic();

@Output(Const_E)
KStream<String,StreamClass> outtCTopic();


}

我什至尝试用不同的名称更改每个进程的名称，但似乎没有任何效果。有人可以帮忙吗。

Answer 1

经过多次调试，我意识到我错过了第 3 个侦听器上的 @Component 注释，因此 Spring boot 没有选择它