Node-rdkafka: Issue when using rebalance_cb

[ArtemisMucaj]

Hi,

I was trying to use rebalance_cb: true today and I came across several issues.

The first one is an issue when trying to connect to Kafka. (with version 1.0.0)

this._client.assign(assignments); ^ Error: Need to specify an array of partitions at Error (native) at KafkaConsumer.assign (/home/.../node_modules/node-rdkafka/lib/kafka-consumer.js:220:16) at KafkaConsumer.conf.rebalance_cb (/home/.../node_modules/node-rdkafka/lib/kafka-consumer.js:57:14)

Second issue (with version 0.7.x)

this._client.unassign(); ^ Error: KafkaConsumer is disconnected at Error (native) at KafkaConsumer.unassign (/home/.../node-rdkafka/lib/kafka-consumer.js:231:16) at KafkaConsumer.conf.rebalance_cb (/home/.../node_modules/node-rdkafka/lib/kafka-consumer.js:59:14)

Any idea?

[webmakersteve]

This may be a bug since I changed the API a bit in a recent commit. I will investigate it.

Thanks for the report.

[pmidge]

@ArtemisMucaj I hit the 0.7.x error, it's a race condition between an earlier call to disconnect and the rebalance callback firing. i wrapped the unassign() in try...catch and ran that build in production for months without issue - no adverse effects observed. only remedy was to NOT handle rebalances, which i was kind of doing - logging the assignments then returning false to allow node-rdkafka to do it.

i'm also hitting the error you report against 1.0.0, but i'm using 1.0.1. the error occurs because the assignments value is undefined in the callback. i am debugging it right now and have not made progress on a solution. i need to move off 0.7.x so i'll likely do without the rebalance handler for now and monitor the issue here.

[ArtemisMucaj]

@pmidge Yup I had it happen in the same context in 0.7.x! I don't really need to run it in production yet but thanks for the tip!

In 1.0.0
https://github.com/Blizzard/node-rdkafka/blob/master/lib/kafka-consumer.js#L56

Changing self.assign(assignment); by self.assign(this.assignments()) seems to work!
No idea why assignment is undefined though. I haven't really had the time to investigate.

[pdykes]

I'm running into this as well: a) assignment is null entering the function, and b, when I try to set it, e.g.: /* have not found a precise example, my take after looking at code usage in the library /
assignment = [];
assignment[0] = {
topic: 'sometopic',
partition: 1,
offset: 0
}; Causes node rdkafka library to core dump: Begin consuming messages from topic: sometopic
Partition Setting: 1
assignment: [ { topic: 'sometopic', partition: 1, offset: 0 } ]
stack smashing detected *: node terminated
Aborted (core dumped)

[webmakersteve]

This is causing you to core dump? I don't think that's the same issue then. Can you post a larger code example?

[webmakersteve]

Fixed with https://github.com/Blizzard/node-rdkafka/commit/6dfa3ae142d6edbb9cc26bf1ecd0f378cc28f378

[hugebdu]

@webmakersteve Stephen, any plans to publish v1.0.5?

Thanks!

[hugebdu]

@webmakersteve Stephen, still on this issue.
I tried with v1.0.5 and now I see the following:

• rebalance_cb is called as expected
• once I call KafkaConsumer#disconnect, the rebalance callback fails on consumer.unassign() with error Error: KafkaConsumer is disconnected and I never get the disconnect callback/event, however seems like the consumer thread is still running since the node process doesn't go down.

[pmidge]

I saw this behavior on earlier builds as well.

My primary motivation for calling unassign was to do a clean shutdown of the consumer process, I found it to be unnecessary. After a ton of experimentation and looking at the network traffic under different scenarios, the brokers will rebalance after the consumer disconnects, and the consumer flushes pending commits during disconnect.

Since unassign requires a connection the error is legitimate, but it's a sort of catch-22. So unless your scenario involves shutting the consumer down but keeping the process alive (in which case unsubscribe would probably be a better solution), IMHO it is superfluous.

[webmakersteve]

Just to clarify on what assignment is:

Assignment is an action done locally to your consumer to tell it what partitions and topics to make fetch calls to. A side effect of rebalancing is that your consumer assigns itself partitions, but assigning and rebalancing are related indirectly.

When you disconnect, if you were to call unassign prior, the broker would never know of your valiant effort to clean up after yourself! No network request is made, and the other consumers in the group are never given the notice that you did that. It is done only for your own benefit - to stop consuming potentially a little earlier.

I can change the default rebalance callback to catch those errors so they do not hold you up, as I would recommend in your own code.

Otherwise, I'm not sure what is stopping you from shutting down. If you enable librdkafka level debug logging it will be possible to see if any librdkafka actions are still being made. You can do this with the debug: fetch,cgrp config parameter and by listening to the event.log event.

[webmakersteve]

Sorry about the 1.0.5 release not being pushed - I think it may have went to our local NPM and not the outside NPM :(

In any event, I've published 1.0.6 which has the changes of 1.0.5. Also, the default rebalance callback (when opted in) try...catches assign and unassign when in a shutdown state. Hopefully that helps with at least the throwing issue.

[hugebdu]

@webmakersteve Steve, thanks for clarification.
So I also did try...catch.
I actually went deeper into the hole with lldb and found out that this loop spins forever once I have the setup I've described.

How do I enable rdkafka logging you mentioned?

[hugebdu]

Ok, found out how to enable logging. Thanks! :)

[hugebdu]

Steve, so this is what I see in logs, and it is stuck forever:

event.log {"severity":7,"fac":"CGRPTERM","message":"[thrd:main]: Group \"group.id_GHcHyIKC\": waiting for rebalance_cb, 1 toppar(s), 0 unassignment(s), 0 commit(s), wait-unassign flag, (state up, join-state wait-revoke-rebalance_cb) before terminating"}

[hugebdu]

Also might this one be our problem?
Our need for rebalance_cb is only for our tests - we need verify broker is up and running before executing our tests so we rely on that callback. We don't need it in production since kafka cluster is up and running. Do you have any idea we could run our is_alive check without the rebalance_cb?

[pipefail]

I think I have reproduced this exact case. @hugebdu @webmakersteve

Here is a Gist whose first case reproduces this contention between disconnect and the rebalance_cb.

While assign and unassign of course are just local calls, subscribe and unsubscribe are not (correct me if I'm wrong). It appears that calling unsubscribe is enough to inform the group manager that the consumer is leaving the group and start the process of the broker marking the consumer as failed. The rebalance_cb for partition revocation is then triggered shortly after disconnect (and unsubscribe) have been called.

The above Gist, when run, will hang forever after the unsubscribe and disconnect calls and the disconnected event will never fire. The broker logs show the consumer being properly marked as failed and dropped from the CG.

Hope this helps. To run the other configuration permutations, just swap/uncomment one the of the other calls in the run method to conditionally omit the rebalance_cb and/or subscribe calls.

[webmakersteve]

I believe that with this commit the issue may be helped:

https://github.com/Blizzard/node-rdkafka/commit/22c713b2d9dfbb2b0ef9d42bf2b890c4ed9e4133

This will allow assigns and unassigns to go through even when the producer isn't connected, hopefully letting the rebalance callback resolve when a custom one is specified.

@petermelias you are correct. Subscribe and unsubscribe require the group leader to re-coordinate the group, just as would happen if one of the consumers crashed.

Going to close this because I think it is fixed in the latest build. Reopen if I'm wrong!