春云流 卡夫卡流 绑定器 卡夫卡异常：无法启动流：'listener'不能为空

Question

我是Kafka Streams和Spring Cloud Stream的新手，但在将与集成相关的代码移动到属性文件中方面，我读过很多关于它的好东西，因此开发人员可以主要关注业务逻辑方面。

这是我的简单应用程序类。

package com.some.events.consumer
import com.some.events.SomeEvent
import org.apache.kafka.streams.kstream.KStream
import org.springframework.boot.autoconfigure.SpringBootApplication
import org.springframework.boot.runApplication
import org.springframework.context.annotation.Bean
import java.util.function.Consumer
@SpringBootApplication
class ConsumerApplication {
@Bean
fun consume(): Consumer<KStream<String, SomeEvent>> {
return Consumer { input -> input.foreach { key, value -> println("Key: $key, value: $value") } }
}
}
fun main(args: Array<String>) {
runApplication<ConsumerApplication>(*args)
}

我的application.yml文件如下。

spring:
cloud:
function:
definition: consume
stream:
bindings:
consume-in-0:
destination: "some-event"
group: "some-event"

我在build.gradle.kts中的依赖项定义如下(此处仅包括相关的依赖项(。

extra["springCloudVersion"] = "2020.0.2"
dependencies {
implementation("org.jetbrains.kotlin:kotlin-reflect")
implementation("org.jetbrains.kotlin:kotlin-stdlib-jdk8")
implementation("org.springframework.cloud:spring-cloud-stream")
implementation("org.springframework.cloud:spring-cloud-stream-binder-kafka-streams")
testImplementation("org.springframework.boot:spring-boot-starter-test")
}
dependencyManagement {
imports {
mavenBom("org.springframework.cloud:spring-cloud-dependencies:${property("springCloudVersion")}")
}
}

当我运行应用程序时，我得到了以下异常。

org.springframework.context.ApplicationContextException: Failed to start bean 'streamsBuilderFactoryManager'; nested exception is org.springframework.kafka.KafkaException: Could not start stream: ; nested exception is java.lang.IllegalArgumentException: 'listener' cannot be null
at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:181) ~[spring-context-5.3.5.jar:5.3.5]
at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:54) ~[spring-context-5.3.5.jar:5.3.5]
at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:356) ~[spring-context-5.3.5.jar:5.3.5]
at java.base/java.lang.Iterable.forEach(Iterable.java:75) ~[na:na]
at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:155) ~[spring-context-5.3.5.jar:5.3.5]
at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:123) ~[spring-context-5.3.5.jar:5.3.5]
at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:935) ~[spring-context-5.3.5.jar:5.3.5]
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:586) ~[spring-context-5.3.5.jar:5.3.5]
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:769) ~[spring-boot-2.4.4.jar:2.4.4]
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:761) ~[spring-boot-2.4.4.jar:2.4.4]
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:426) ~[spring-boot-2.4.4.jar:2.4.4]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:326) ~[spring-boot-2.4.4.jar:2.4.4]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1313) ~[spring-boot-2.4.4.jar:2.4.4]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1302) ~[spring-boot-2.4.4.jar:2.4.4]
at com.some.events.consumer.ConsumerApplicationKt.main(ConsumerApplication.kt:22) ~[main/:na]
Caused by: org.springframework.kafka.KafkaException: Could not start stream: ; nested exception is java.lang.IllegalArgumentException: 'listener' cannot be null
at org.springframework.cloud.stream.binder.kafka.streams.StreamsBuilderFactoryManager.start(StreamsBuilderFactoryManager.java:94) ~[spring-cloud-stream-binder-kafka-streams-3.1.2.jar:3.1.2]
at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:178) ~[spring-context-5.3.5.jar:5.3.5]
... 14 common frames omitted
Caused by: java.lang.IllegalArgumentException: 'listener' cannot be null
at org.springframework.util.Assert.notNull(Assert.java:201) ~[spring-core-5.3.5.jar:5.3.5]
at org.springframework.kafka.config.StreamsBuilderFactoryBean.addListener(StreamsBuilderFactoryBean.java:268) ~[spring-kafka-2.6.7.jar:2.6.7]
at org.springframework.cloud.stream.binder.kafka.streams.StreamsBuilderFactoryManager.start(StreamsBuilderFactoryManager.java:84) ~[spring-cloud-stream-binder-kafka-streams-3.1.2.jar:3.1.2]
... 15 common frames omitted

Process finished with exit code 1

注意，我知道我需要配置Serde和Avro相关的东西(我使用Avro作为事件模式(，但问题是，流甚至不会运行。

有人能给我指正确的方向吗？我试着在谷歌上搜索这个，但没有人发布由"listener"不能为null引起的问题。谢谢

Answer 1

这是一个错误；它被固定在3.1.3.SNAPSHOT 中

https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/commit/f25dbff2b7fc0d0c742dd674a9e392057a34c86d

https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1030#issuecomment-804039087

我不确定那里的评论；将千分尺添加到类路径应该可以解决此问题。

Answer 2

destination: "some-event"应该指向一个kafka主题。像destination: "some-event-topic"。

然后，您必须为侦听器consume-in-0创建一个接口。使用spring注释将使项目加载此侦听器，并且它将不再为null。

import org.apache.kafka.streams.kstream.KStream;
import org.springframework.cloud.stream.annotation.Input;
public interface KafkaListenerBinding {
    @Input("consume-in-0")
    KStream<String, String> inputStream();
}

然后创建一个@Service来处理来自侦听器@StreamListener("consume-in-0")的消息。

import lombok.extern.log4j.Log4j2;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Service;
@Log4j2
@Service
@EnableBinding(KafkaListenerBinding.class)
public class KafkaListenerService {
@StreamListener("consume-in-0")
public void process(KStream<String, String> input) {
input.foreach((k,v) -> log.info(String.format("Key: %s, Value: %s",k,v)));
}
}

注意：尽管@Gary Russel说了这个错误，但我将用实现Spring服务的功能方式来完成我的回答。函数样式可以通过在application.yml文件中定义函数来实现。有一个内部约定，使用函数的名称以及绑定的posfixin-0和out-0。定义绑定时必须使用此选项。更多详细信息请点击此处。

spring:
cloud:
stream:
function:
definition: transformToUpperCase
bindings:
transformToUpperCase-in-0:
destination: input-func-topic
transformToUpperCase-out-0:
destination: output-func-topic

然后用@Configuration和@EnableAutoConfiguration注释类，并确保lambda方法与您在application.yml文件中为function.definition定义的方法相同。

@Configuration
@EnableAutoConfiguration
public class KafkaListenerFunctionalService {
@Bean
public Function<KStream<String, String>, KStream<String, String>> transformToUpperCase() {
return input -> input
.peek((k, v) -> log.info("Functional received Input: {}", v))
.mapValues(i -> i.toUpperCase());
}
}

Answer 3

我也有同样的问题，所以首先我添加了io.milter依赖项(从maven安装最新版本(

第二个从SimpleMeterRegistry创建Bean，它解决了问题

@Bean
SimpleMeterRegistry simpleMeterRegistry() {
return new SimpleMeterRegistry();
}