Kafkajs: 5 sec lag with multi broker

[vykimo]

Hi everyone, I noticed a very special behavior of kafkajs.
When I use only one broker I consume immediately messages produced.
But when I have a multi broker node, I get systematically a 5 sec lag between production and consumption.

I tried with kafka console in order to know if it is a bad configuration but no, produced messages are immediately written into consummer console.
Maybe you have already heard of similar issue.
Let me know.

[tulios]

Without a lot of investigation, I can say that the default maxWaitTime is 5s, the fetch request will block for maxWaitTime or minBytes, so I am guessing that when the topic doesn't have new messages and you publish it will still wait for 5s (since the brokers have to sync due to acks=-1 by default). On the next request, the minBytes rule will be satisfied, and the data will return right away.

[vykimo]

So this behavior doesn't happens with one broker because sync isn't required, is it?

Then, to fix it I have to change acks value and/or lower maxWaitTime and minBytes?

[vykimo]

Hmm so, minBytes is set by default to 1 (my messages are always greater than 1byte) and acks is mandatory set to -1 to guarantee idempotency.
So the only parameter I am able to change is maxWaitTime right?
If I set it to 100ms, is it too low or it is ok?

[vykimo]

So it's working with 100ms for maxWaitTime.
But I have 3 questions :

1. Why when I have only one broker, messages are immediately read and when I have multi-brokers they wait for maxWaitTime (I confirmed this behavior again. With same messages + 1, then 2 brokers. bug?)
2. What are the consequences to drop maxWaitTime from 5000 to 100ms ?
3. Where can I set maxWaitTime in a constructor or something (I force it in kafkajs code currently) ?

[tulios]

Hey @vykimo, sorry for the delay.

Where can I set maxWaitTime in a constructor or something (I force it in kafkajs code currently) ?

You can configure the maxWaitTime when creating the consumer https://kafka.js.org/docs/consuming#a-name-options-a-options

The option is called maxWaitTimeInMs, e.g:

kafka.consumer({ groupId: 'test', maxWaitTimeInMs: 100 })

What are the consequences to drop maxWaitTime from 5000 to 100ms ?

You just add more pressure to Kafka (more requests over-time), but that's ok. You have to optimize your consumers for batch or "real-time", where real-time means low minBytes and smaller wait time.

Why when I have only one broker, messages are immediately read and when I have multi-brokers they wait for maxWaitTime (I confirmed this behavior again. With same messages + 1, then 2 brokers. bug?)

Are you using kafka-console with acks=-1 and idempotency? I would have to check what kind of configurations this tool has by default because it might be optimized for developer experience (which makes sense), but not quite for a production environment. But I don't know; I have to check it out.

[vykimo]

Yes, I am using acks by default (-1) and idempotency.

You have to optimize your consumers for batch or "real-time", where real-time means low minBytes and smaller wait time.

That's my main problem, I would prefer to lower minBytes but it's already at 1 by default, however fetch seems not to be triggered when message is published __or__ maybe the lag is due to acks = -1 (wait ack of all replicas) ?... @tulios

[bobzsj87]

I observed the same issue that testing on a single broker env has no lag at all. But on 2 brokers, you can observe random lags from 0-5 seconds (using default consumer params). Using kafka-console-consumer gets immediate result back.

[bobzsj87]

fetch.max.wait.ms in kafka doc https://kafka.apache.org/documentation/ is 500, maybe the default should be set to 500?

[bobzsj87]

Just to add a bit more information. in our live env, for a topic with partition =3, replication=2, we can observe the lag. But when partition=1, replication=2, the lag is not observable.

[vykimo]

For me, it's an issue of minBytes kafkajs-side or fetch.min.bytes Kafka Broker side...
Because it is never triggered with multi-broker. Consumer waits for expiration of maxWaitTimeInMs.

[bobzsj87]

To summarize my case:
1) single broker, multi-partition = ok
2) multi-broker, single-partition = ok
3) multi-broker, multi-partition = lag

yes, @vykimo the minBytes is not triggered in case 3)

[bobzsj87]

@tulios could this be a bug?

[tulios]

@vykimo @bobzsj87 it could be a bug, but we run many high load consumers, so since we have a lot of machines in those groups, we already expect some delay when rebalancing. Summer is approaching in Sweden, and we are wrapping up some previous work, so I apologize for the delayed answers, we are a bit spread thin since a couple of weeks. I will find some time to investigate this soon, but if someone can start digging it would help a lot.

[bobzsj87]

@tulios thanks for the reply. I have done some isolated test with fresh install of Kafka server and can reproduce the issue, please find the detailed results and test scripts here: https://gist.github.com/bobzsj87/4971386799677e9840653aaba87679bf

[tulios]

I think I know what is happening. Kafka uses the pull model, so the consumers don't receive any push messages from the server, so this is what is happening:

1) consumer A joins the group
2) the group is empty, so consumer A it becomes the leader and starts consuming right away
3) consumer B joins the group
4) Kafka asks the group to rebalance and waits for the leader to assign partitions
5) consumer A doesn't know that the group is rebalancing, it will discover that when calling Kafka and getting an error back. This error will most likely happen on the next heartbeat, which by default is 3s
6) consumer A discovers the rebalance and reassigns the partitions
7) the group rebalances

So why 5s, each fetch by default will wait for 5s or 1 byte, so I am guessing that it is the combination of the heartbeat time + some wait time on the fetch.

@bobzsj87 can you try reducing the heartbeatInterval and maxWaitTimeInMs to 500ms?

[bobzsj87]

@tulios but there is only one group in my test. No rebalancing happening.

[IvanovOleg]

I have the same issue. I have 3 brokers, two topics (1 partition, two partitions), one producer and one consumer. If I use topic with a single partition, everything works fast. If I use >= 2 partitions, I have 5 sec lag on consumer. maxWaitTimeInMs = 100 fixes the issue for me

[vykimo]

@IvanovOleg Yes me too, but that's not a fix but an overexploitation of the server (one call every 100ms instead of one call only when broker has enough message bits)

[bobzsj87]

@tulios I guess I found the bug after examing the code. It is due to the fetch mechanism of the consumerGroup (https://github.com/tulios/kafkajs/blob/master/src/consumer/consumerGroup.js), that it chooses the broker where the replica leader is on, then fetch each broker in parallel using Promise.all(). However, under low traffic, in a multiple broker setup, one broker A which is the leader of partition 1 gets a new message, in the same time, the other broker B, which is the leader of partition 2 is idling. In this case, the kafka broker immediately return the messages from broker A, but keep broker B waiting until maxWaitTime. Since it is promise.all, the fetch() will wait until maxWaitTime.

Not sure if I am correct but if that's true, then the batch mechanism is flawed and should be redesigned to address per-broker parallelism.

[vykimo]

Oh well seen @bobzsj87

[IvanovOleg]

Any updates?

[bobzsj87]

Any updates?

I switched to kafka-node

[tulios]

Sorry for the delay, we had quite an unusual summer.

@bobzsj87 I think your explanation makes sense, I checked kafka-node and they make the requests in series, so the approach might work better when you have brokers idling. I think the model is correct, but I will investigate if we can change some things to make the parallel requests yield the results instead of waiting for all the requests to finish. This change should address the issue.

So, something like:

        |---> A --> yield -----------------------------------|
fetch - |---> B ----------------------------> yield  --------| - end of fetch loop
        |---> C ------------------------------------> yield -|

consumerGroup.fetch can return a generator, so the runner knows how many requests are expected and can feed the batches as they come to the callbacks and finish once every request has arrived.

[tulios]

I managed to write the async generator to yield the results over time, so I think I can update the fetch loop to use the generator. The only problem is that this feature competes with partitionsConsumedConcurrently, which requires the data to be available. I will discuss this with @Nevon, but I think the new implementation has benefits over the concurrently option.

[Nevon]

We had a discussion regarding this change in the concurrency model, and it looks like it will work perfectly fine with the partitionsConsumedConcurrently functionality, as they take care of different aspects. This makes it so that the requests themselves yield their results before all have finished, and partitionsConsumedConcurrently makes sure that the userland code is only called with a concurrency of n - regardless of when the request results are yielded.

I think this change makes a lot of sense.

[tulios]

We have a branch with the new code working, we will refactor some of that next week and give it a try

[tulios]

1.13.0-beta.0 is out with this change