I am trying to consume messages from a topic in Kafka via a simple Spring boot application via the spring-cloud-stream-binder-kafka-streams library.
I am following the documentation here - https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/blob/master/docs/src/main/asciidoc/kafka-streams.adoc
My code is as below: package co.zip.poc.springbootkafkakotlin
import org.springframework.boot.autoconfigure.SpringBootApplication
import org.springframework.boot.runApplication
import org.apache.kafka.streams.kstream.KStream
import org.springframework.context.annotation.Bean
import java.util.function.Consumer
@SpringBootApplication
class SpringBootKafkaConsumer {
@Bean
fun process(): Consumer<KStream<Any, Any>> {
println("=======================> Creating a consumer for reading stuff<=====================")
return Consumer { input -> input.foreach { key, value ->
println("============key = $key")
println("===========value = $value")
}}
}
}
fun main(args: Array<String>) {
runApplication<SpringBootKafkaConsumer>(*args)
}
application.yml
spring:
application:
name: customer-balance
cloud:
stream:
kafka:
streams:
binder:
configuration:
application:
id: customer-balance
bindings:
process_in:
destination: "dbserver1.inventory.customers"
logging:
level:
org.springframework.kafka.config: trace
I have my local kafka running on port 9092. I am able to consume messages via the kafka console consumer -
kafka-console-consumer.sh \
--bootstrap-server localhost:9092 \
--from-beginning \
--property print.key=true \
--topic dbserver1.inventory.customers
The above produces 4 messages on the console. But when I run the spring boot app, I don't get anything printed for the key = .. and value = print messages.
What am I doing wrong?
