Kafkajs: Support for max.poll.records

[mt4c]

Will there be support for max.poll.records in config so that I can control how many records are fetched by consumer?

[tulios]

Hi @mt4c, we had a discussion some time ago about this property and concluded that it wasn't the time to invest in this feature for a couple of reasons.

1) If you have a stable message, it's quite simple to have the average message size, for example, we do it like this:

const AVG_MESSAGE_SIZE = 280 // bytes
const NUMBER_OF_MESSAGES_TO_RECEIVE = 400
const MAX_BYTES_PER_PARTITION = NUMBER_OF_MESSAGES_TO_RECEIVE * AVG_MESSAGE_SIZE
const MAX_BYTES = 32 * MAX_BYTES_PER_PARTITION

2) If you use plain JSON and have variable sizes, or have way too different sizes, it becomes a harder problem. The messages might have 100 bytes or 53 MB, which makes the size detection harder and we would need to change the maxBytes configuration dynamically, so we think userland code would be easier to write and deal with this case for now. You can use eachBatch and only feed the expected amount of messages to your handlers.

[JaapRood]

@tulios regarding maxBytes being able to change dynamically: that _would_ be one of the main requirements to implement a proper _Stream_ in userland, which is quite possibly the cleanest and most platform native way to consume messages in a memory-efficient and performant way. Feeding a stream from an eachBatch is what we're doing currently, but you're basically creating a second buffer that you don't really have any control over. Now there's more required to implement a _Stream_ on top of this (like #417), but it's worth to keep in mind as a limiting factor of eachBatch.

[tulios]

Yeah, I don't think it's a lot of work to allow changes to maxBytes, I am talking about having the logic to increase/decrease based on the messages in KafkaJS, more as a priority thing. If we get more time or people interested, we can definitely do it.

NOTE: Over here we had to upgrade the cluster and make significant changes to work the growth we had, so most of my time is now debugging some mysterious bugs and helping the teams to adjust to the new setup.

[mt4c]

OK, after some search on the google I found that it is a bad idea to implment task queues on kafka. Trying to find another way. So should I close this issue?

[JaapRood]

Not sure what in Kafka makes it a bad idea for implementing task queues, we're definitely employing it in such a way. One thing is that it's basically a lower level tool than most dedicated task queues, so it's not as easy to work that way, but it's got all the primitives you need. That probably means that if all you're doing is a task queue, and you're not _that_ worried about things like solid processing guarantees (like at-least-once or exactly-once), it might be a bit overkill for just that.

[tulios]

I guess you need to be creative to have priority and other mechanisms with Kafka, is feasible, and we have a similar setup, but depending on the needs it can be overkill, especially due to the increase in complexity.

[nickyromeijn]

Are there any plans to add this feature to the roadmap? I'm currently running into a situation where it would be very nice to have. While creating new consumers and setting maxBytes dynamically is technically possible. It would mean I'd have to know and broadcast the message size in advance to consumers. Which seems a bit weird considering the premise of Kafka.

In my situation I use Kafka as task queue and data bus and want consumers to each pick up one task, process it (~ takes a few minutes) and retrieve the next.

[Nevon]

No one is actively working on this, and it's fairly unlikely that one of us will have time to pick it up.

Regarding your particular use-case, why does it matter if the consumer picks up 1 or 10 messages at a time? It can still process them one at a time, and anyway it's not likely anyone else can pick up a message on the same partition, so it's not like you're gaining any concurrency.

[nickyromeijn]

Actually I am and would be. I work with massively different message sizes and sometimes processing can take up to 10+ minutes. In that case I add in extra consumers which are AWS EC2 machines to pick up the slack. If one machine pulls for example 500 messages I do not gain anything by spinning up more machines. My workaround for now is just to create topics with 100 partitions which allows me to scale up to 100 consumers and let the rebalancer do its work.

Love the lib though. Might pick the feature up myself if i have time.

[SEUNGGWANSHIN]

Hi @mt4c, we had a discussion some time ago about this property and concluded that it wasn't the time to invest in this feature for a couple of reasons.

1. If you have a stable message, it's quite simple to have the average message size, for example, we do it like this:

const AVG_MESSAGE_SIZE = 280 // bytes
const NUMBER_OF_MESSAGES_TO_RECEIVE = 400
const MAX_BYTES_PER_PARTITION = NUMBER_OF_MESSAGES_TO_RECEIVE * AVG_MESSAGE_SIZE
const MAX_BYTES = 32 * MAX_BYTES_PER_PARTITION

1. If you use plain JSON and have variable sizes, or have way too different sizes, it becomes a harder problem. The messages might have 100 bytes or 53 MB, which makes the size detection harder and we would need to change the maxBytes configuration dynamically, so we think userland code would be easier to write and deal with this case for now. You can use eachBatch and only feed the expected amount of messages to your handlers.

ㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡㅡ
now... that time
i wanna use 'max.poll.records' options.
have you plan to develop 'max.poll.records' options?