Application doesn't finish gracefully after client.disconnect

[Pchelolo]

I expect that after a producer/consumer is disconnected it should destroy all RdKafka objects and let the application exit, however it doesn't happen.

Code example:

const Kafka = require('node-rdkafka');
const producer = new Kafka.Producer({
    'metadata.broker.list': 'localhost:9092'
});
producer.connect(undefined, () => {
    producer.produce({
        topic: 'test_dc.resource_change',
        message: 'test'
    }, () => {
        producer.disconnect(() => {
            console.log('Disconnected');
        });
    });
});

Versions: kafkaOS X El Capitan
Expected behaviour: application sends one message to the topic and shuts down.
Actual behaviour: application sends one message to the topic but continues running.

A bit of investigation: gdb shows that we have 8 threads running after disconnect was called:

  Id   Target Id         Frame 
* 1    Thread 0x1203 of process 55925 0x00007fff8f611eca in kevent () from /usr/lib/system/libsystem_kernel.dylib
  2    Thread 0x1303 of process 55925 0x00007fff8f60afae in semaphore_wait_trap () from /usr/lib/system/libsystem_kernel.dylib
  3    Thread 0x1403 of process 55925 0x00007fff8f60afae in semaphore_wait_trap () from /usr/lib/system/libsystem_kernel.dylib
  4    Thread 0x1503 of process 55925 0x00007fff8f60afae in semaphore_wait_trap () from /usr/lib/system/libsystem_kernel.dylib
  5    Thread 0x1603 of process 55925 0x00007fff8f60afae in semaphore_wait_trap () from /usr/lib/system/libsystem_kernel.dylib
  6    Thread 0x1703 of process 55925 0x00007fff8f610db6 in __psynch_cvwait () from /usr/lib/system/libsystem_kernel.dylib
  7    Thread 0x1803 of process 55925 0x00007fff8f61107a in select$DARWIN_EXTSN () from /usr/lib/system/libsystem_kernel.dylib
  8    Thread 0x1903 of process 55925 0x00007fff8f61107a in select$DARWIN_EXTSN () from /usr/lib/system/libsystem_kernel.dylib

Threads 1-5 are normal, so whatever prevents shutdown is in threads 6-8, I suspect it's on thread 6, here's the backtrace:

#0  0x00007fff8f610db6 in __psynch_cvwait () from /usr/lib/system/libsystem_kernel.dylib
#1  0x00007fff9cd3f728 in _pthread_cond_wait () from /usr/lib/system/libsystem_pthread.dylib
#2  0x000000010079630b in uv_cond_wait ()
#3  0x000000010078a2ab in worker ()
#4  0x0000000100796000 in uv.thread_start ()
#5  0x00007fff9cd3e99d in _pthread_body () from /usr/lib/system/libsystem_pthread.dylib
#6  0x00007fff9cd3e91a in _pthread_start () from /usr/lib/system/libsystem_pthread.dylib
#7  0x00007fff9cd3c351 in thread_start () from /usr/lib/system/libsystem_pthread.dylib
#8  0x0000000000000000 in ?? ()

Any ideas where could the source of a problem be?

[webmakersteve]

The only thing I can think that is doing this is that client->close() actually has an absurdly long (for general node use-cases) default timeout. Has console.log reported the word 'Disconnected' or is even that blocked?

The reason I ask is that we have a test-case that is running a pretty similar piece of code: https://github.com/Blizzard/node-rdkafka/blob/master/e2e/producer.spec.js#L113 which is passing CI.

[Pchelolo]

The only thing I can think that is doing this is that client->close() actually has an absurdly long (for general node use-cases) default timeout.

The 'disconnect' was reported on the console, so the callback is called, and it's called pretty fast.

The reason I ask is that we have a test-case that is running a pretty similar piece of code: https://github.com/Blizzard/node-rdkafka/blob/master/e2e/producer.spec.js#L113 which is passing CI.

I think the reason why it's passing is because you are explicitly calling process.exit(0) here, however that shouldn't be a requirement for the node app to shutdown, right?

For us it's pretty important since we run tests in mocha and different tests require different configurations, so we heavily use disconnect and then recreate the clients with different configs.

[webmakersteve]

Has console.log('Disconnected'); executed in your example before it hangs?

[Pchelolo]

Oh, sorry, wrong indentation on my previous comment..

Has console.log('Disconnected'); executed in your example before it hangs?

The 'disconnect' was reported on the console, so the callback is called, and it's called pretty fast.

[webmakersteve]

Looks like this happens regardless of whether a message is produced or not. In fact, just instantiating the class makes it do this.

The only RdKafka objects created at that phase are the configs and their requisite callbacks. I can try to play around and see if storing Persistent v8 handles of objects rather than rdkafka configs until connect time fixes it.

[webmakersteve]

Was able to identify it as the Dispatcher class.

[webmakersteve]

New branch with a fix in place that should make it shutdown gracefully: https://github.com/Blizzard/node-rdkafka/tree/graceful-shutdown. Currently running some tests on it but feel free to try it and let me know if that fixed the problem for you.

[Pchelolo]

@webmakersteve It definitely makes things different, but:

This test made it coredump:

"use strict";
var kafka = require('./lib');
var consumer = new kafka.KafkaConsumer({
    'metadata.broker.list': 'localhost:9092',
    'group.id': 'test'
});
consumer.connect(undefined, function() {
    console.log('connected');
    consumer.subscribe([ 'test_dc.resource_change' ]);
    consumer.disconnect();
});

But after I've updated the branch and rebuilt it once again I can't reproduce any more, but here's a dump (a relevant part of it). Just FYI, if it rings some bells

Crashed Thread:        0  Dispatch queue: com.apple.main-thread

Exception Type:        EXC_CRASH (SIGABRT)
Exception Codes:       0x0000000000000000, 0x0000000000000000

Application Specific Information:
Assertion failed: (0), function uv__finish_close, file ../deps/uv/src/unix/core.c, line 264.


Thread 0 Crashed:: Dispatch queue: com.apple.main-thread
0   libsystem_kernel.dylib          0x00007fff8f610f06 __pthread_kill + 10
1   libsystem_pthread.dylib         0x00007fff9cd414ec pthread_kill + 90
2   libsystem_c.dylib               0x00007fff9e9cb6e7 abort + 129
3   libsystem_c.dylib               0x00007fff9e992df8 __assert_rtn + 321
4   node                            0x0000000105d7d342 uv_run + 680
5   node                            0x0000000105c4f9eb node::Start(int, char**) + 594
6   node                            0x0000000105485d34 start + 52

[webmakersteve]

Can you try one more time. I noticed that RIGHT after i said it and updated the branch. If you re-pulled afterwards it would make sense that it didn't happen again.

uv_close does not expect the memory to be freed so I had to make an adjustment.

[Pchelolo]

@webmakersteve checked again, no EXC_CRASH any more, but there's a test that still doesn't work:

"use strict";
var kafka = require('./lib');
var consumer = new kafka.KafkaConsumer({
    'metadata.broker.list': 'localhost:9092',
    'group.id': 'test'
}, {
    'auto.offset.reset': 'smallest'
});
consumer.connect(undefined, function() {
    console.log('connected');
    consumer.subscribe([ 'test_dc.resource_change' ]);
    consumer.consume(function (err, msg) {
        console.log(err, msg);
        console.log('Calling disconnect');
        consumer.disconnect(function(err, info) {
            console.log('Diconnected', err, info);
        });
    });
});

Important thing is that I have some messages in the topic. So for some reason after I consume something, disconnect never happens and the callback is never called. I've tried waiting for like 5 minutes on this one, I know that RdKafka::Handle::close might be super slow, but not as slow.

GDB shows some interesting stuff - we have 8 threads, as always thread 1 is event loop, 2-4 are libuv thread pull, threads 6 and 7 are in tread_join() and thread 8 is in poll(). In my experience that happened when some reference to an RdKafka::Message leaked or wasn't deleted before disconnecting. That object contains a reference to RdKafka::KafkaConsumer. It has some custom reference counting mechanism embedded, so having more some undeleted RdKafka::Message objects before calling RdKafka::KafkaConsumer::Close make the latter call stuck. But that's just an unproved theory. In support for it, if I run the same code against an empty topic, so I get timeout instead of a message, disconnect happens properly

[webmakersteve]

In this case, does it log 'Disconnected'? If it does, it means client->close() has completed execution and it would be the event loop still having open threads.

https://github.com/Blizzard/node-rdkafka/blob/master/src/message.cc#L103

The message is freed as the node::Buffer free callback, so it should only happen when node garbage collects. This was done to prevent needing to copy the data from Kafka to save time.

[Pchelolo]

@webmakersteve Nope, in this case it doesn't log Disconnected, so client->close() never finishes.

The message is freed as the node::Buffer free callback, so it should only happen when node garbage collects. This was done to prevent needing to copy the data from Kafka to save time.

So, I guess that's the root cause of this problem... And there's not much we can do I guess - performance is way more critical then graceful shutdown. Lemme play with it a bit