Node-rdkafka: Limited number of consumers per process

[Pchelolo]

Since the driver uses Nan::AsyncQueueWorker for background job scheduling, we end up using built-in libuv thread pool. That means, that by default there's only 4 threads in the pool, and this number couldn't be increased more then to 128 threads.

Each consumer in the flowing mode submits a ConsumerConsumeLoop job, which is blocking, so it occupies one background thread completely - this means that number of consumers per process is limited to 4 (3 actually, since we need to save at least 1 thread from the pool for other tasks)

Another possible way to use up all the threads in the pool is to call consume(cb) in non-flowing mode many-many times on a topic without any messages coming - each call would create a ConsumerConsume work that will occupy the thread from the pool until the message is there, so it will block all the other operations which could be going on (producing, metadata requests etc etc etc).

I'm wondering if you think this might be a problem? In our use-case we don't fork a worker per consumer group, but instead create all of the consumers in every worker process and rely on kafka rebalancing for assigning individual partitions to workers since it's easier and better for failover (every worker is replaceable but any other one)

Do you think this will hit you too at some point? I've created this issue mostly to get the understanding on your thoughts on this.

[webmakersteve]

It's definitely something I've run into before in my testing, but in our particular uses cases it wasn't something impactful.

We have architected our systems to have a master process that reads off of all the Kafka Topics we care about. Worker processes listen over an IPC channel and the master sends them serialized data to process. When they are done processing the data, they send periodic reports to master so it can aggregate the work that has been done and convert it into commits and keep track of how busy each worker is.

I definitely see the appeal in changing this - we don't want libuv to be over-allocated and block the rest of the process from queuing events.

As for the additional possible way at using all the threads in the pool: consume should time-out periodically and release the thread. At least, that is the intended behavior. Are you observing something different?

[Pchelolo]

As for the additional possible way at using all the threads in the pool: consume should time-out periodically and release the thread. At least, that is the intended behavior. Are you observing something different?

Ye, that happens, but unfortunately doesn't help in my use-case, will have to redesign this part a bit to make it use flowing mode - then I'd need to occupy only 1 background thread per consumer - that might help to push the needle a bit further.

Do you think that you will address this problem generically at some point? Are you interested in a contribution that would reimplement Nan::AsyncQueueWorker to avoid using internal libuv thread pool and use a custom thread pool instead (with unlimited capacity and lazy initialisation)?

[webmakersteve]

I don't find any reason why the alternative you suggest wouldn't be better than the current implementation, so we would definitely be interested in contributions that remove this as an issue for users.

[InfinitiesLoop]

That'd be awesome. Good performance is important to the library, so I'd love to see that a change like this leaves its performance either unchanged or better (aside from being more scalable). There are scripts in the bench folder to help with measuring.

[jdconley]

How would one go about increasing the Threadpool size of libuv in this context? I've just run into this issue when writing some integration tests.

[webmakersteve]

Before queueing any asynchronous work, set process.env.UV_THREADPOOL_SIZE to a value at least 2 greater than the number of consumers you plan to have running concurrently.

[jdconley]

Thank you @webmakersteve ! That unclogged my pipes. Much easier than rewriting my tests to use IPC. 👍

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs.