将kstream与spring cloud stream和kafka 1.0.1结合使用

vmdwslir  于 2021-06-07  发布在  Kafka
关注(0)|答案(2)|浏览(429)

我试图创建一个简单的程序来打印Kafka主题的kstream。我经常得到一个npe，完全没有想法。
我已经使用了SpringCloud流绑定器kafka streams dependency，我正在使用最新版本的SpringCloud“finchley.m9”。
我写的代码是：

@Component
@EnableBinding(KafkaStreamsProcessor.class)
public class EventListener{

    @StreamListener("input")
    public void listen(KStream<String,String> kstream){
        kstream.print();
    }
}

application.properties具有：

spring.cloud.stream.bindings.input.destination=slot-events
spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.binder.keySerde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.binder.valueSerde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000
spring.cloud.stream.kafka.streams.binder.configuration.applicationId=listener
spring.cloud.stream.kafka.streams.binder.configuration.zookeeper.connect=localhost:2181

当我启动服务时，控制台上不断出现以下错误：

018-03-31 22:57:52.641  INFO 26301 --- [           main] sStreamListenerSetupMethodOrchestrator$1 :  values:
        application.id = default
        application.server =
        bootstrap.servers = [localhost:9092]
        buffered.records.per.partition = 1000
        cache.max.bytes.buffering = 10485760
        client.id =
        commit.interval.ms = 1000
        connections.max.idle.ms = 540000
        default.deserialization.exception.handler = class org.apache.kafka.streams.errors.LogAndFailExceptionHandler
        default.key.serde = class org.apache.kafka.common.serialization.Serdes$StringSerde
        default.timestamp.extractor = class org.apache.kafka.streams.processor.FailOnInvalidTimestamp
        default.value.serde = class org.apache.kafka.common.serialization.Serdes$StringSerde
        key.serde = class org.apache.kafka.common.serialization.Serdes$StringSerde
        metadata.max.age.ms = 300000
        metric.reporters = []
        metrics.num.samples = 2
        metrics.recording.level = INFO
        metrics.sample.window.ms = 30000
        num.standby.replicas = 0
        num.stream.threads = 1
        partition.grouper = class org.apache.kafka.streams.processor.DefaultPartitionGrouper
        poll.ms = 100
        processing.guarantee = at_least_once
        receive.buffer.bytes = 32768
        reconnect.backoff.max.ms = 1000
        reconnect.backoff.ms = 50
        replication.factor = 1
        request.timeout.ms = 40000
        retry.backoff.ms = 100
        rocksdb.config.setter = null
        security.protocol = PLAINTEXT
        send.buffer.bytes = 131072
        state.cleanup.delay.ms = 600000
        state.dir = /tmp/kafka-streams
        timestamp.extractor = null
        value.serde = class org.apache.kafka.common.serialization.Serdes$StringSerde
        windowstore.changelog.additional.retention.ms = 86400000
        zookeeper.connect = localhost:2181

2018-03-31 22:57:52.656  INFO 26301 --- [           main] o.apache.catalina.core.StandardService   : Stopping service [Tomcat]
2018-03-31 22:57:52.685  INFO 26301 --- [           main] ConditionEvaluationReportLoggingListener :

Error starting ApplicationContext. To display the conditions report re-run your application with 'debug' enabled.
2018-03-31 22:57:52.693 ERROR 26301 --- [           main] o.s.boot.SpringApplication               : Application run failed

java.lang.IllegalStateException: java.lang.NullPointerException
        at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.adaptAndRetrieveInboundArguments(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:273) ~[spring-cloud-stream-binder-kafka-streams-2.0.0.RC3.jar!/:2.0.0.RC3]
        at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.orchestrateStreamListenerSetupMethod(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:154) ~[spring-cloud-stream-binder-kafka-streams-2.0.0.RC3.jar!/:2.0.0.RC3]
        at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.doPostProcess(StreamListenerAnnotationBeanPostProcessor.java:195) ~[spring-cloud-stream-2.0.0.RC3.jar!/:2.0.0.RC3]
        at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.lambda$postProcessAfterInitialization$0(StreamListenerAnnotationBeanPostProcessor.java:167) ~[spring-cloud-stream-2.0.0.RC3.jar!/:2.0.0.RC3]
        at java.lang.Iterable.forEach(Iterable.java:75) ~[na:1.8.0_161]
        at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.injectAndPostProcessDependencies(StreamListenerAnnotationBeanPostProcessor.java:285) ~[spring-cloud-stream-2.0.0.RC3.jar!/:2.0.0.RC3]
        at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.afterSingletonsInstantiated(StreamListenerAnnotationBeanPostProcessor.java:105) ~[spring-cloud-stream-2.0.0.RC3.jar!/:2.0.0.RC3]
        at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:777) ~[spring-beans-5.0.4.RELEASE.jar!/:5.0.4.RELEASE]
        at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:868) ~[spring-context-5.0.4.RELEASE.jar!/:5.0.4.RELEASE]
        at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:549) ~[spring-context-5.0.4.RELEASE.jar!/:5.0.4.RELEASE]
        at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:140) ~[spring-boot-2.0.0.RELEASE.jar!/:2.0.0.RELEASE]
        at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:752) [spring-boot-2.0.0.RELEASE.jar!/:2.0.0.RELEASE]
        at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:388) [spring-boot-2.0.0.RELEASE.jar!/:2.0.0.RELEASE]
        at org.springframework.boot.SpringApplication.run(SpringApplication.java:327) [spring-boot-2.0.0.RELEASE.jar!/:2.0.0.RELEASE]
        at org.springframework.boot.SpringApplication.run(SpringApplication.java:1246) [spring-boot-2.0.0.RELEASE.jar!/:2.0.0.RELEASE]
        at org.springframework.boot.SpringApplication.run(SpringApplication.java:1234) [spring-boot-2.0.0.RELEASE.jar!/:2.0.0.RELEASE]
        at dg.athena.sideprojects.kafkastreampoc.KafkastreampocApplication.main(KafkastreampocApplication.java:15) [classes!/:na]
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_161]
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_161]
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_161]
        at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_161]
        at org.springframework.boot.loader.MainMethodRunner.run(MainMethodRunner.java:48) [kafkastreampoc-0.0.1-SNAPSHOT.jar:na]
        at org.springframework.boot.loader.Launcher.launch(Launcher.java:87) [kafkastreampoc-0.0.1-SNAPSHOT.jar:na]
        at org.springframework.boot.loader.Launcher.launch(Launcher.java:50) [kafkastreampoc-0.0.1-SNAPSHOT.jar:na]
        at org.springframework.boot.loader.JarLauncher.main(JarLauncher.java:51) [kafkastreampoc-0.0.1-SNAPSHOT.jar:na]
Caused by: java.lang.NullPointerException: null
        at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.getkStream(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:294) ~[spring-cloud-stream-binder-kafka-streams-2.0.0.RC3.jar!/:2.0.0.RC3]
        at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.adaptAndRetrieveInboundArguments(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:235) ~[spring-cloud-stream-binder-kafka-streams-2.0.0.RC3.jar!/:2.0.0.RC3]
        ... 24 common frames omitted

有人能建议一下吗？

apache-kafkaapache-kafka-streamsspring-cloud-stream

2条答案

按热度按时间
ejk8hzay

ejk8hzay1#

这是在最新快照中修复的错误。能否尝试将活页夹升级到1.0.0.build-snapshot并重试？

赞(0）分享  回复(0）举报  2021-06-07
9rbhqvlz

9rbhqvlz2#

当使用springstarter或springinitializer时，包含springcloudstream包含gradle（或maven）依赖项

testImplementation 'org.springframework.cloud: spring-cloud-stream-test-support'

每次执行“bootrun”启动应用程序时，都会生成npe或非法状态异常。
当注解掉上面的gradle依赖时，npe消失了，一切正常。我们已经用springcloudstream和kafka streams在6个不同的spring启动应用程序上重复了这一点。
目前假设一切正常，因为代码在消除测试依赖的情况下运行良好。但是，千万不要不了解问题就继续进行（因为他们以后可能会再次出没）。
如有任何想法/解决方法，我们将不胜感激。

赞(0）分享  回复(0）举报  2021-06-07
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com