Concurrent message processing

[Nevon]

This implements a simple way to achieve partition-aware concurrency. This allows a consumer to consume messages from multiple partitions concurrently, while still ensuring ordering within each partition.

Usage

consumer.run({
  partitionsConsumedConcurrently: 3, // Default: 1
  eachMessage: async ({ topic, partition, message }) => {
      // This will be called up to 3 times concurrently
  }
})

In order to benefit from this kind of concurrency, the consumer needs to be subscribed to a topic with at least 2 partitions, and the work being done in eachMessage/eachBatch needs to be async (preferably I/O). I added a guideline in the docs to avoid people setting this to Math.MAX_SAFE_INTEGER and expecting their consumer to be super fast.

This is implemented by creating a kind of queue every time we finish a fetch. Each item in the queue is processed concurrently up to the specified limit. If one throws, we reject all the pending items. I did this to behave similarly to the existing behavior where we bail out of the loop on failure, but given that each item is a separate topic/partition, this is something we might be able to revisit in order to keep processing other partitions even if one of them fails.

Fixes #330.

[Nevon]

This should shift instead, to grab from the head of the list

[Nevon]

Fixed in 7069059

[tulios]

is this right? I think it is backwards right, partitionsConsumedConcurrently should be lower than partitions

[Nevon]

I rewrote this to actually make sense.

[t-d-d]

Is the number of CPU cores relevant for a Node application? This is still single threaded processing.

[Nevon]

It depends on your workload. If the message processing is entirely synchronous, then this setting is completely irrelevant as all the work takes place on the main thread and will thus be executed sequentially regardless.

However, in most cases the work is not entirely synchronous. In many cases, we call remote APIs or read/write to a database, etc. In those cases, some of the work takes place outside of the main thread, such as when doing network I/O. If each message requires you to make a network call to some API, you would normally just sit around and wait while the remote service is doing some work, but with this concurrency model you can instead fire off multiple of these requests concurrently.

So then you might say that you would just set the concurrency limit to whatever number of partitions you have, in order to achieve maximum concurrency, but that may not actually increase throughput for multiple reasons. First, your downstream dependency might not appreciate getting that many concurrent requests and it's likely their performance may suffer. Even if that's fine, network or disk I/O is not infinitely parallelisable, and correlates with the number of cores you have.

While they're talking about connection pool sizing in this article, the principles are very much the same, and it's a really good high-level explanation: https://github.com/brettwooldridge/HikariCP/wiki/About-Pool-Sizing

So the complex answer is that it really depends on your workload and your dependencies. The recommendation is pretty conservative, because otherwise people will put a gigantic number in there thinking that "more is better" and end up getting lower throughput and potentially really bursty behavior. Throughput may very well increase with higher concurrency, given the right conditions, but it's better to start low and increase it over time while measuring the effect until you hit the sweet spot.

[t-d-d]

Can I set it to a high number to always process all partitions in parallel? Is the behaviour the same for any setting > # consumed partitions ?

[Nevon]

Yes. You can set it to a number higher than the number of partitions, and the effective result will be the same as if you set it to the number of partitions.

[tulios]

Fixes #320