Spring Boot and Kotlin: Dynamic Kafka topic listeners

Viewed 130

I am writing a service with Sprint Boot and Kotlin. The service should listen to Kafka topics. I am using Spring Cloud Stream for that.

What's working is having a hardcoded topic to listen to. For example in application.yaml I define the topic to listen:

spring:
  cloud:
    stream:
      default:
        contentType: application/*+avro
        group: my-consumer-group
        consumer:
          useNativeDecoding: false
      # Binding-specific configs (Kafka agnostic)
      bindings:
        my_topic:
          # Topic to consume from
          destination: my_topic

and then access it like:

interface MyTopicSink {

    @Input(INPUT)
    fun input(): SubscribableChannel

    companion object {
        const val INPUT = "my_topic" # (From `application.yaml`)
    }
}

/**
 * Reads from my_topic Kafka topics.
 */
@Service
@EnableBinding(MyTopicSink::class)
class MyFancyConsumer() {
    /**
     * Listens & consumes from the my_topic.
     */
    @StreamListener(MyTopicSink.INPUT)
    fun processTopic(
        @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) key: String,
        @Payload payload: org.apache.avro.generic.GenericData.Record
    ) {
          log.info("Yaay, this is working")
      }
}

That's great and all, but does anyone know if I can make this dynamic? For example, one of my ideas is to have a list of topics defined somewhere, and then just have this consumer be dynamic. Reading from all the topics. Anyone did something similar? Thoughts?

0 Answers
Related