Kafkajs: Consumer: Fails to subscribe after run()

[denic]

Hi guys, I have a follow up question to https://github.com/tulios/kafkajs/issues/336#issuecomment-490830101

I started to play around with kafkajs and tried to integrate it to our high-level solution. We need to dynamically subscribe to topics during runtime.
To me it looks like subscribing to topics after consumer.run(...) was called does not add the new topic to the list of topics that get fetched in the ConsumerGroup:

('topic-test' and 'service__alpha' gets subscribed before calling run(), 'XXX' after run())

Is it me doing it wrong, is the consumer not intended to be used in this way or did I stumble over a bug?

Thanks a lot in advance!

[tulios]

Hi @denic, looking at the code I can say that this won't work with the current implementation. To add topics on demand, the client will have to force a rebalance so that the leader can assign the partitions of the new topics.

We will need to change some internals to make this happen, what you can do in the meantime is to use the admin client to fetch the list of topics continuously and once you find a new topic you restart the consumer, which will use the regexp subscribe to fetch the new topics correctly.

We are planning a small release soon, but I don't think we would include this change now.

[denic]

Thanks @tulios . Much appreciated!

[tulios]

Feel free to re-open the issue; I can create a feature request issue later describing the steps.

[Arkh1]

@tulios - Could you elaborate on what you mean by restarting the consumer? Like the OP I too have a need for dynamically creating and subscribing to topics. I tried calling consumer.stop() followed by consumer.run(), but that didn't seem to help. I also tried just creating a new instance of the same consumer and subscribing to all existing topics again. That also didn't seem to work.

Also, has there been any updates on supporting dynamic topics in a more natural way?