enable.auto.commit: false and commit()

[killfill]

Hi!

I was trying to explicit use commit(), like this:

var stream = consumer.getReadStream(config.kafka_topic, {
    fetchSize: 10
});

stream.on('data', msgs => {

    msgs.map(msg => {

        console.log('commiting...')
        consumer.commit(msg, err => {
            console.log('   commited!', err)
        })

    })
})

Noticed that the 'commited' callback, takes quite some time to be called (2-3 seconds for each one of them)

Is it correct to send commits like this?
Is so, why would it be so slow?

Thanks!!

[webmakersteve]

It shouldn't take that long, but librdkafka does batch commits together to be processed. I believe the default time is much less than 3 seconds. However, how long it takes could be determined by when the call actually gets processed by libuv which may depend on how many outstanding tasks you have in the node event loop.

Can you try running it with a higher value for UV_THREADPOOL_SIZE? It defaults to 4 so try setting it higher and see if things work faster. If that's the case, it looks like you have too many events queued and you'll probably want to consolidate things.

In your example, you are queueing quite a few things in the event loop which may slow the process down. I would recommend you instead commit a TopicPartition.

Instead of providing a message to the callback, you can provide the greatest offset for each partition in each topic. e.g.

var stream = consumer.getReadStream(config.kafka_topic, {
    fetchSize: 10
});

stream.on('data', msgs => {
    // Judging from your example, you're reading from one topic
    var topicPartitions = {};


    msgs.forEach(msg => {
        if (!topicPartitions.hasOwnProperty(msg.partition)) {
            topicPartitions[msg.partition] = msg.offset;
        } else {
            var currentOffset = topicPartitions[msg.partition];
            if (currentOffset < msg.offset) {
                 topicPartition[msg.partition] = msg.offset;
            }
        }
    });

    for (var x in topicPartitions) {
        console.log('Committing');
        consumer.commit({ topic: config.kafka_topic, partition: x, offset: topicPartitions[x] }, err => {
            console.log('   commited!', err)
        });
    }
})

I haven't tested it, but that may do the trick.

Nonetheless I can take a look to see if this is something internal top the library or not.

[killfill]

Interesting, It seems that the commits are getting send somehow in groups of 4.

Doing your trick does help!

Thanks!

[webmakersteve]

We are taking a look at this and will potentially move the threads we use out of the default threadpool to avoid this potential issue, but I'm glad (at least, for now) this fixed the issue for you! :)

You can see part of the discussion in Issue #4