Kafkajs: Fetch partitions in parallel

[hmagarotto]

When the processing of one partition is slow, all partitions are affected.
It's because the "fetch" for other partitions only is performed after the process completion for all partitions.
The "fetch" for all topics and partitions are synchronous.
So, it's not possible to use the partitions as a unit of parallelism.

Can we deal with this using some setting or using any different strategy?

Test with a slow process (exaggerated) on partition 0:
https://github.com/hmagarotto/kafkajs-test-partition-parallelism

[ankon]

This sounds a bit related to #370 (#570) to me. If so, maybe you could try the beta releases with the test?

[hmagarotto]

Hi, thanks for your response.

What I understood about this refactoring is that:

The consumer will process batches immediately after a response for each fetch request.
But, a new fetch for this consumer group will only be started after all batches has already been processed. For a new fetch start, we need to wait for "enqueuedTasks" on "Runner::fetch".

So, if I have understood correctly, a slow task for some partition will block new fetches for all other partitions.

I tried the test with the new beta version and all partitions stay blocked by slower partition task.

Test code updated with beta KafkaJS: https://github.com/hmagarotto/kafkajs-test-partition-parallelism

Similar needs for another library/language: https://github.com/akka/alpakka-kafka/issues/110

[Nevon]

Your understanding is correct, @hmagarotto. Currently the fetch loop will wait until all partitions have been processed before issuing new fetch requests. I've been having the same discussion on our Slack today, and I think it makes a lot of sense to start the next fetch for a broker as soon as we have processed the last response from it, to avoid having to wait for the slowest one.

[tristan-redox]

@Nevon, we recently tried a BETA version of KafkaJS and noticed that another fetch started before the current processing fetch completed. The problem we encountered was that the second fetch received the same message/offset from the same topic/partition that was currently being processed in the yet to finish fetch. This resulted in the same message being processed twice "concurrently" within <50 ms of each other. This seems like a bug, should I raise a bug report against the BETA version?

There are a number of concerns around starting the next fetch before the previous finishes. For example, preventing the behavior above (duplicate concurrent processing of the same message) or guaranteeing FIFO processing for each topic/partition a consumer is assigned. If this feature is added, I believe you would need to ensure the second fetch does not fetch from topic/partitions that are still being processed by the current unfinished fetch. This seems like it could be difficult to ensure.

[Nevon]

we recently tried a BETA version of KafkaJS and noticed that another fetch started before the current processing fetch completed. The problem we encountered was that the second fetch received the same message/offset from the same topic/partition that was currently being processed in the yet to finish fetch. This resulted in the same message being processed twice "concurrently" within <50 ms of each other. This seems like a bug, should I raise a bug report against the BETA version?

Please do. Especially if you can produce a failing test or at least a repro case for this.

To be clear, we would never start a fetch for a partition for which we haven't processed all messages in the current batch yet, as partitions are the unit of concurrency in Kafka. However, there's theoretically nothing stopping us from issuing a new fetch request for partition 1 just because we haven't finished processing the messages for partition 2 yet.

[sparcs360]

Hi all - are there any plans to work on this?

Regarding workarounds:

1. Lower maxWaitTimeInMs.
2. Create slow consumers on separate clients (would that work? or do they end up being pooled?)
3. Any more?

[sparcs360]

workaround 2 gives me No active topic partitions, sleeping for 5000ms - I can't make sense of that...

If I put each consumer in it's own consumer group then it works (although I prefer workaround 1)

[ankon]

workaround 2 gives me No active topic partitions, sleeping for 5000ms - I can't make sense of that...

The 5000 is simply the maxWaitTime you configured, and the message is logged when the consumer is not subscribed to any topics, _or didn't get any partitions to fetch assigned to it_. In that case KafkaJS just waits that time before issuing another fetch attempt, in the hope that until then maybe the assignment has changed.

Check whether you subscribed to the correct topics in all your consumers.