Node-rdkafka: Consumer stop consuming after unknown idle time.

[zaryab19aug]

After an unknown idle time in which no new message is received to consumer, the kafka consumer stopped consuming messages. Altough consumer.connectedTime() gives the connected time and consumer.isConnected() also returns true but messages are not received untill the service is restarted.

No broker transport failure or any other type of error is occurred, everything seems to working fine.

Environment Information

• OS: Linux
• Node Version: 10.15.3
• NPM Version: 6.4.1
• node-rdkafka version: 2.7.4

Consumer configurations

    'client.id': config.name,
    'metadata.broker.list': ['localhost:9092'],
    'heartbeat.interval.ms': 5000,
    'socket.keepalive.enable': true,
    'fetch.wait.max.ms': 100,
    'message.max.bytes': 1000,
    'enable.auto.commit': false,
    'group.id': `${name}_consumer`,

[dapobelieve]

@zaryab19aug were you able to fix this issue? I've been having it for weeks now and I couldn't find any solution

[rognorak]

I am also seeing this issue, with 2.8.1. Not sure if its relevant, @zaryab19aug and @dapobelieve , are either of you using AWS MSK as your brokers ?

[dapobelieve]

No I use digitalocean, But on my machine I'm using docker

[Ben56]

I also encountered the same problem

[iradul]

Please try with the latest version 2.9.0. There was a regression in librdkafka 1.3.0 that is fixed in 1.4.2:

Fix produce/consume hang after partition goes away and comes back,
such as when a topic is deleted and re-created (regression in v1.3.0).

Also having only one broker in production setup is not recommended. You should have at least 3 brokers.

[DanielFrag]

I had the same problem here (version 2.9.0) using the confluent cloud.
We (me and my team) noticed that, when this lack of connection occurs, an error is passed to consumer and the method consumer.assignments returns an empty array.
Probably, this is not the best solution, but seems to work.
We check the error object to trigger a "reconnect task".

// Something like this
const checkError = (errorData) => errorData.origin === 'local' && errorData.message === 'broker transport failure' && errorData.code === -1;
const reconnect = (consumer) => consumer.disconnect(() => consumer.connect());

On Flowing mode this error comes as the first argument of the handler, passed to consumer.consume(handler) method.
On Non Flowing mode this error comes on event event.error.
We also check the consumer's assignments before to start the reconnect task.

[alagerkvist]

I faced the same issue as @DanielFrag (version 2.9.0) on IBM Cloud. Will it be an update soon regarding this problem?

Thanks in advance!

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs.