Programatically adding kafka consumers

[ckosmowski]

The use of @KafkaListener annotation is very convenient if we use the micronaut-kafka api directly. However if we want so provide our own classes that offer pre-configured kafka consumers we would have to implement the poll loop ourselfs allthough all the beautiful code is already there in KafkaConsumerProcessor.

I'd like to suggest a feature where we would be able to add a preconfigured kafkaConsumer to KafkaConsumerProcessor that reuses the code that is already there in the process() method but takes the configuration that is handed as parameters instead as the annotation information.

As a more specific example we'd like to be able to have a KafkaListener where offsetStrategies and all the parameters are configured, also the KafkaListenerException handler is set properly. The User (i.E. Developers of other teams) should then be able to just specify a topic, a groupId and how they'd like to process the messages. They shouln't care for committing or resetting offsets in case of an Exception or such things. This would be a great benefit for us.

[graemerocher]

Sounds like a nice idea. PRs are of course welcome 🙏

[ckosmowski]

If "PRs" is short for prayers i'll send you some. If you mean pull requests this feature might actually be a good one to start contributing. I'll do my very best.

[mcac0006]

I've been using micronaut-kafka for 2 years but this is the first time I would be contributing to the code base. Nonetheless, I would like to lend a hand.

However if we want so provide our own classes that offer pre-configured kafka consumers we would have to implement the poll loop ourselfs allthough all the beautiful code is already there in KafkaConsumerProcessor.

This would help - do you happen to have a code snippet please?

[awattez]

hello, I find this feature very useful in a reactive context, it is the exact definition of reactive.

Let's imagine creating consumers on the fly in order to react to a message or other... do you have any news or leads for implementing programmatic consumers that will be accessible through the ConsumerRegistry and other useful components of micronaut-kafka?