Kafkajs: Issue when consumer more one partition

[moix80]

When I have a consumer with more that one partition consuming the performance is really bad, around 3 seconds for consuming mensajes. I tried with other lib and the performance is good.

[ankon]

Can you provide a _bit_ more details of what you're doing, show some code, and maybe describe your expectations? Remember that other libraries have different defaults that target different use cases (think latency/throughput goals).

[moix80]

Hi Ankon, I have identify the issue with more detail. So, when I have a consumer joined to partitions of different leaders the latency is around 3 seconds. However if partitions have the same leader the performance is around 4ms, for exaple:

I have topic_demo_kafka with 6 partitions, if I have a consumer joined in partitions 0,3 de latency is really good. But if I have the same consumer joined to partitions 0,2,4 the latency for 4 is ok but the latency for 0,2 is around 3 seconds (really high).

    Topic: topic_demo_kafka Partition: 0    Leader: 1   Replicas: 1,2,3 Isr: 1,2,3
Topic: topic_demo_kafka Partition: 1    Leader: 3   Replicas: 3,1,2 Isr: 3,1,2
Topic: topic_demo_kafka Partition: 2    Leader: 2   Replicas: 2,3,1 Isr: 2,3,1
Topic: topic_demo_kafka Partition: 3    Leader: 1   Replicas: 1,3,2 Isr: 1,3,2
Topic: topic_demo_kafka Partition: 4    Leader: 3   Replicas: 3,2,1 Isr: 3,2,1
Topic: topic_demo_kafka Partition: 5    Leader: 2   Replicas: 2,1,3 Isr: 2,1,3

My question, it's a normal behavior in Kafka or it's a issue in kafka.js ?

[moix80]

Any update here? I cant reproduce issue in Java. So I think the issue is about kafkajs.

[ankon]

I don't know anything from the top of my head that would produce such a behavior.

Can you post a self-standing executable example so that we can possibly try to reproduce this issue?

[moix80]

Let me know if you need my help for reproduce the issue
This example use a topic with 6 partitions and replication factor 3
One producer and 2 consumers

==========================================================

Create topics

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
clientId: 'my-app',
brokers: YOUR_BROKER,
ssl:{rejectUnauthorized: false}
});
const admin = kafka.admin();

async function main(){
await admin.connect();
await admin.createTopics({topics:[{topic:'topic-test',replicationFactor:3,numPartitions:6}]} );
await admin.disconnect();
}

main();

======================================================

CONSUMERS

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
clientId: Date.now().toString(),
brokers: YOUR_BROKER,
ssl:{rejectUnauthorized: false}
});

main().then(()=>console.log('finished')).catch((err)=>console.log(err));

async function main(){
const consumer1 = kafka.consumer({ groupId: 'grouptest'})
await consumer1.connect();
await consumer1.subscribe({ topic: 'topic-test', fromBeginning: true});
await consumer1.connect();

const consumer2 = kafka.consumer({ groupId: 'grouptest'})
await consumer2.connect();
await consumer2.subscribe({ topic: 'topic-test', fromBeginning: true});
await consumer2.connect();

consumer1.run({
    eachMessage: async ({ topic, partition, message }) => {
        let t1 = Date.now();
        let time = (t1-JSON.parse(message.value.toString('utf8')).time);
        let producer = JSON.parse(message.value.toString('utf8')).producer;
        let key = message.key.toString('utf8');
        console.log(time,topic,partition,producer,key);

    },
});

consumer2.run({
eachMessage: async ({ topic, partition, message }) => {
let t1 = Date.now();
let time = (t1-JSON.parse(message.value.toString('utf8')).time);
let producer = JSON.parse(message.value.toString('utf8')).producer;
let key = message.key.toString('utf8');
console.log(time,topic,partition,producer,key);

    },
});

}

=====================================================

PRODUCER

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
clientId: 'my-app',
brokers: YOUR_BROKER,
ssl:{rejectUnauthorized: false}
});
var names = ["A","B","C","D","E","F","G","H","I","J","K","L","M","N"];

function getRandomInt(min, max) {
return Math.floor(Math.random() * (max - min)) + min;
}
async function main(){
let name = names[getRandomInt(0, names.length-1)];
let producer = kafka.producer();
//let partition = getRandomInt(0,5);
await producer.connect();
var cont = 0;
var t = Date.now();
setInterval(async function(){
cont=cont+1;
console.log(partition,cont);

    await producer.send({
        topic: 'topic-test',

        messages: [
            { key: cont.toString(), value: JSON.stringify({producer:name,time:Date.now()}) },
        ],
    });
},1000);

}

main();

[silversoul93]

We have the same issue in a production environment.
We have a topic with 6 partitions and the KafkaJS consumers are so slow; in the test environment we have same topic but with 1 partition and the consumers can consume messages in about 20ms instead of 3 seconds.

We have a consumer written in kotlin too, but it consumes messages very fast, the problem is probably related to KafkaJS.

We use the standard eachMessage configurations, we have not custom behaviors except for the message commit.

Can you support us?

Could the problem be related to the partitionAwareConcurrency descripted in your documentation?

[silversoul93]

This is our situation in production environment:

Topic: commands PartitionCount: 6 ReplicationFactor: 3
          Topic          | Partition | Leader | Replicas |   ISR
+------------------------+-----------+--------+----------+---------+
  commands               |         0 |      3 | [3 5 4]  | [5 4 3]
  commands               |         1 |      8 | [8 7 3]  | [8 7 3]
  commands               |         2 |      7 | [7 8 6]  | [8 7 6]
  commands               |         3 |      1 | [1 0 5]  | [5 1 0]
  commands               |         4 |      2 | [2 1 7]  | [7 2 1]
  commands               |         5 |      4 | [4 2 8]  | [8 4 2]

Same for the other topic, 6 partitions with different leaders, and the consumption time is around 3 seconds or more.
We really need to fix this.

Thanks

[ankon]

Say, would you be able to compare this with a "prerelease version"?

I'm wondering a bit whether this is a variant of #370, which has been fixed -- but that fix isn't released yet in a stable version.

[silversoul93]

Say, would you be able to compare this with a "prerelease version"?

We tried using the prerelease version (kafkajs@beta) and we noted a very little performances improvement (from 11/15 seconds for a complete flow to 8/10 seconds, not enough anyway).

I'm wondering a bit whether this is a variant of #370, which has been fixed -- but that fix isn't released yet in a stable version.

We fix the problem by setting the maxWaitTime parameter to 100 milliseconds.

I hope that with next stable release we will not need this workaround.

Thanks for the support ❤️

[Nevon]

We fix the problem by setting the maxWaitTime parameter to 100 milliseconds.

I hope that with next stable release we will not need this workaround.

This is not a workaround, nor a bug. This is the intended behavior in Kafka, and works more or less the same way in all Kafka clients (although the exact default settings may differ).

Basically, when a fetch request is made to a broker, if there aren't enough messages available to fill up to maxBytes/maxBytesPerPartition, the broker will block and not respond for up to maxWaitTime milliseconds to see if there are more messages coming in. Without this, in a situation with a low amount of data, the client would end up making crazy amounts of requests since the broker would immediately respond with nothing, causing the client to poll very fast. This comes at the cost of latency, where as you noticed, if there are only a few messages available on the partition, you won't receive those messages for up to maxWaitTime milliseconds.

In version 1.12.0, this is a bit worse because we issue requests to all brokers simultaneously (up to a limit) and then wait for all of them to respond before we start processing messages. So if you have uneven partitions, such that one partition has messages and another on a different broker doesn't, you would have to wait up to maxWaitTime milliseconds on every fetch, even though one of the brokers has already responded with messages. In the beta version, we changed this so that as soon as a broker responds, we will start processing the messages, even before other brokers have responded, which mitigates this a bit.