Delay Kafka consumer initialization on application startup

[mvmandapati]

I am migrating an old thorntail app to Quarkus + SmallRye Kafka. It consumes messages from multiple channels, process them and send to third party vendor.

During startup(StartupEvent) app prefetches some data from vendor which is required for processing messages. Challenge I am having is coordinating data prefetching and Kafka consumer start-up. Consumer is starting much before and as a result message processing is failing.

I have looked at the documentation and stackoverflow with not much luck. Is it possible to implement above usecase with Quarkus without using Native Kafka client?

I have also checked #27205 but not sure if its solves above usecase.

Related discussions on StackOverflow:

https://stackoverflow.com/questions/62824824/is-it-possible-to-delay-consuming-from-channels-amqp-kafka-etc-in-quarkus
https://stackoverflow.com/questions/62329274/how-to-deploy-a-kafka-consumer-being-in-pause-mode-until-i-signal-to-start-consu

[quarkus-bot[bot]]

/cc @alesj, @cescoffier, @ozangunalp

[ozangunalp]

Still a work in progress, but I think when this will be merged smallrye/smallrye-reactive-messaging#1860 it can answer your question.

[jponge]

We have runSubscriptionOn to offload to an executor. Otherwise .plug allows plugging a custom operator, that sounds doable to have one that delays forwarding a subscription request according to some policy of your own.

[cristianovagos]

Sorry for digging this up, but is there an example of this working @ozangunalp @jponge ?
My use case is to delay the subscription/consumption until a StartupEvent or some custom event inside the application is fired.

[magro]

@cristianovagos Did you find something you could share?

[cristianovagos]

@cristianovagos Did you find something you could share?

Yes. I used a combination of KafkaConsumerRebalanceListener and health checks.
Since it is called on the polling thread, if something is wrong on the callback we can pause/resume consumers.

[magro]

@cristianovagos Thanks for sharing! 👍

[gkozyryatskyy]

Hello guys! Is there any workable solution for this?
In my case i need to delay consumption of @Incoming while StartupEvent will not finished.
Thank you!

[scspijker]

We also have a use case for this, we want to have the consumer disabled on startup, and startup the consumer on custom signals.

[seseso908]

Would the pausable feature be a good fit to your requirements?
https://smallrye.io/smallrye-reactive-messaging/latest/concepts/pausable-channels/

[mvmandapati]

I am assuming pausable channel works for simple consumers without outgoing channel as well.

Also when channel is initially paused, is the underlying topic subscribed but not polled?