Spring Cloud Kafka Stream Multibinder BeanCreationException: Error creating bean with name 'kafkaStreamsFunctionProcessorInvoker'

Viewed 32

I am trying to create a spring boot application using spring cloud kafka stream which reads input from kafka cluster 1 and send it to kafka cluster 2 using single kafkastream application.

I am getting following exception during startup.

***************************
APPLICATION FAILED TO START
***************************

Description:

A component required a bean named 'sessionBrokers-KafkaStreamsBinderConfigurationProperties' that could not be found.


Action:

Consider defining a bean named 'sessionBrokers-KafkaStreamsBinderConfigurationProperties' in your configuration.

My configuration is as below

spring.cloud.function.definition=itemConsumedStream

#Input Configuration
spring.cloud.stream.bindings.itemConsumedStream-in-0.binder=sessionBrokers
spring.cloud.stream.bindings.itemConsumedStream-in-0.destination=${session.event.data.input.topic}
spring.cloud.stream.kafka.bindings.itemConsumedStream-in-0.consumer.configuration.key.deserializer=org.apache.kafka.common.serialization.StringSerializer
spring.cloud.stream.kafka.bindings.itemConsumedStream-in-0.consumer.configuration.value.deserializer=in.custom.JsonNodeDeserializer

#Output Configuration
spring.cloud.stream.bindings.itemConsumedStream-out-0.binder=searchBrokers
spring.cloud.stream.bindings.itemConsumedStream-out-0.destination=${itemConsumed.events.output.topic}
spring.cloud.stream.kafka.bindings.itemConsumedStream-out-0.producer.configuration.key.serializer=org.apache.kafka.common.serialization.StringSerializer
spring.cloud.stream.kafka.bindings.itemConsumedStream-out-0.producer.configuration.value.serializer=in.custom.JsonSerializer


#Input Binder Configurations
spring.cloud.stream.binders.sessionBrokers.type=kafka
spring.cloud.stream.binders.sessionBrokers.environment.spring.cloud.stream.kafka.streams.binder.brokers=${session.event.data.input.kafka.brokers}

#Output Binder Configurations
spring.cloud.stream.binders.searchBrokers.type=kafka
spring.cloud.stream.binders.searchBrokers.environment.spring.cloud.stream.kafka.binder.brokers=${search.kafka.brokers}

pom dependencies are spring boot version

2.7.1

spring cloud version

2021.0.0
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>

This is working if I use single binder ie

spring.cloud.stream.kafka.streams.binder.brokers=${session.event.data.input.kafka.brokers}

instead of multiple binders

0 Answers
Related