threads not getting cleaned up #1803

vk-coder · 2018-05-11T20:12:46Z

Read the FAQ first: https://github.com/edenhill/librdkafka/wiki/FAQ

Description

If broker communication is not established threads don't get cleaned up

How to reproduce

Create producer with broker ip which is not yet started.
Similarly create consumer with broker ip which is not yet started.
fetch metadata using consumer->metadata , which would return error
delete producer and consumer
repeat from beginning

IMPORTANT: Always try to reproduce the issue on the latest released version (see https://github.com/edenhill/librdkafka/releases), if it can't be reproduced on the latest version the issue has been fixed.

Checklist

IMPORTANT: We will close issues where the checklist has not been completed.

Please provide the following information:

librdkafka version (release number or git tag): 0.11.0
Apache Kafka version: <REPLACE with e.g., 0.10.2.3>
librdkafka client configuration:
Producer config
conf->set("metadata.broker.list", brokers, errstr);
conf->set("dr_cb", delivery_cb, errstr);
conf->set("event_cb", event_cb, errstr);
conf->set("log_level", "0", errstr);
conf->set("client.id", m_kafka_client_id, errstr);

consumer config
conf->set("metadata.broker.list", brokers, errstr);
conf->set("bootstrap.servers", brokers, errstr);
conf->set("topic.metadata.refresh.interval.ms","3000", errstr);
conf->set("default_topic_conf", tconf.get(), errstr);
conf->set("log_level", "0", errstr);
conf->set("group.id", group_id, errstr);
conf->set("client.id", m_kafka_client_id, errstr);
conf->set("auto.offset.reset", "earliest", errstr);
conf->set("rebalance_cb", rebalance_cb, errstr);
conf->set("statistics.interval.ms", "3000", errstr);
conf->set("event_cb", event_cb, errstr);

Operating system: CentOS Linux release 7.4.1708 (Core)
Provide logs (with debug=.. as necessary) from librdkafka
LOG-7-WAKEUPFD: [thrd:app]: 10.60.11.3:9092/bootstrap: Enabled low-latency partition queue wake-ups
LOG-7-BRKMAIN: [thrd::0/internal]: :0/internal: Enter main broker thread
LOG-7-WAKEUPFD: [thrd:app]: 10.60.11.3:9092/bootstrap: Enabled low-latency ops queue wake-ups
LOG-7-STATE: [thrd::0/internal]: :0/internal: Broker changed state INIT -> UP
LOG-7-BROADCAST: [thrd::0/internal]: Broadcasting state change
LOG-7-BROKER: [thrd:app]: 10.60.11.3:9092/bootstrap: Added new broker with NodeId -1
LOG-7-TOPIC: [thrd:app]: New local topic: test1-b
LOG-7-BRKMAIN: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Enter main broker thread
LOG-7-TOPPARNEW: [thrd:app]: NEW test1-b [-1] 0x1eb6e10 (at rd_kafka_topic_new0:282)
LOG-7-CONNECT: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: broker in state INIT connecting
LOG-7-METADATA: [thrd:app]: Skipping metadata refresh of 1 topic(s): no usable brokers
LOG-7-TOPIC: [thrd:app]: New local topic: test1-u
LOG-7-TOPPARNEW: [thrd:app]: NEW test1-u [-1] 0x1eb29e0 (at rd_kafka_topic_new0:282)
LOG-7-METADATA: [thrd:app]: Skipping metadata refresh of 1 topic(s): no usable brokers
LOG-7-CONNECT: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Connecting to ipv4#10.60.11.3:9092 (plaintext) with socket 11
LOG-7-STATE: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Broker changed state INIT -> CONNECT
LOG-7-BROADCAST: [thrd:10.60.11.3:9092/bootstrap]: Broadcasting state change
LOG-7-NOINFO: [thrd:main]: Topic test1-b partition count is zero: should refresh metadata
LOG-7-NOINFO: [thrd:main]: Topic test1-u partition count is zero: should refresh metadata
LOG-7-METADATA: [thrd:main]: Skipping metadata refresh of 2 topic(s): no usable brokers
LOG-7-NOINFO: [thrd:main]: Topic test1-b partition count is zero: should refresh metadata
LOG-7-NOINFO: [thrd:main]: Topic test1-u partition count is zero: should refresh metadata
LOG-7-METADATA: [thrd:main]: Skipping metadata refresh of 2 topic(s): no usable brokers
LOG-7-NOINFO: [thrd:main]: Topic test1-b partition count is zero: should refresh metadata
LOG-7-NOINFO: [thrd:main]: Topic test1-u partition count is zero: should refresh metadata
LOG-7-METADATA: [thrd:main]: Skipping metadata refresh of 2 topic(s): no usable brokers
LOG-7-NOINFO: [thrd:main]: Topic test1-b partition count is zero: should refresh metadata
LOG-7-NOINFO: [thrd:main]: Topic test1-u partition count is zero: should refresh metadata
LOG-7-METADATA: [thrd:main]: Skipping metadata refresh of 2 topic(s): no usable brokers
LOG-7-NOINFO: [thrd:main]: Topic test1-b partition count is zero: should refresh metadata
LOG-7-NOINFO: [thrd:main]: Topic test1-u partition count is zero: should refresh metadata
LOG-7-METADATA: [thrd:main]: Skipping metadata refresh of 2 topic(s): no usable brokers
LOG-7-DESTROY: [thrd:app]: Terminating instance
LOG-7-TERMINATE: [thrd:app]: Interrupting timers
LOG-7-TERMINATE: [thrd:app]: Sending TERMINATE to main background thread
LOG-7-BROADCAST: [thrd:app]: Broadcasting state change
LOG-7-TERMINATE: [thrd:app]: Joining main background thread
LOG-7-DESTROY: [thrd:main]: Destroy internal
LOG-7-DESTROY: [thrd:main]: Removing all topics
LOG-7-TOPPARREMOVE: [thrd:main]: Removing toppar test1-b [-1] 0x1eb6e10
LOG-7-DESTROY: [thrd:main]: test1-b [-1]: 0x1eb6e10 DESTROY_FINAL
LOG-7-TOPPARREMOVE: [thrd:main]: Removing toppar test1-u [-1] 0x1eb29e0
LOG-7-DESTROY: [thrd:main]: test1-u [-1]: 0x1eb29e0 DESTROY_FINAL
LOG-7-TERMINATE: [thrd:main]: Purging reply queue
LOG-7-TERMINATE: [thrd:main]: Decommissioning internal broker
LOG-7-TERMINATE: [thrd:main]: Join 2 broker thread(s)
LOG-7-TERMINATE: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Handle is terminating: failed 0 request(s) in retry+outbuf
LOG-7-BROKERFAIL: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: failed: err: Local: Broker handle destroyed: (errno: Operation now in progress)
LOG-7-STATE: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Broker changed state CONNECT -> DOWN
LOG-7-BROADCAST: [thrd:10.60.11.3:9092/bootstrap]: Broadcasting state change
LOG-7-BUFQ: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Purging bufq with 0 buffers
LOG-7-BUFQ: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Updating 0 buffers on connection reset
LOG-7-TERMINATE: [thrd::0/internal]: :0/internal: Handle is terminating: failed 0 request(s) in retry+outbuf
LOG-7-BROKERFAIL: [thrd::0/internal]: :0/internal: failed: err: Local: Broker handle destroyed: (errno: Success)
LOG-7-STATE: [thrd::0/internal]: :0/internal: Broker changed state UP -> DOWN
LOG-7-BROADCAST: [thrd::0/internal]: Broadcasting state change
LOG-7-BUFQ: [thrd::0/internal]: :0/internal: Purging bufq with 0 buffers
LOG-7-BUFQ: [thrd::0/internal]: :0/internal: Updating 0 buffers on connection reset
LOG-7-TERMINATE: [thrd:main]: Main background thread exiting
LOG-7-TERMINATE: [thrd:app]: Destroying op queues
LOG-7-TERMINATE: [thrd:app]: Termination done: freeing resources
Provide broker log excerpts
Critical issue

The text was updated successfully, but these errors were encountered:

edenhill · 2018-05-11T20:16:49Z

Please try to reproduce on latest release v0.11.4

vk-coder · 2018-05-11T20:25:51Z

Ohh i tried it on 0.11.4 as well but mentioned the version that is in our production. sorry about that.

above logs are production and below logs are from 0.11.4 version

LOG-7-WAKEUPFD: [thrd:app]: 10.60.11.3:9092/bootstrap: Enabled low-latency partition queue wake-ups
LOG-7-WAKEUPFD: [thrd:app]: 10.60.11.3:9092/bootstrap: Enabled low-latency ops queue wake-ups
LOG-7-BRKMAIN: [thrd::0/internal]: :0/internal: Enter main broker thread
LOG-7-STATE: [thrd::0/internal]: :0/internal: Broker changed state INIT -> UP
LOG-7-BROADCAST: [thrd::0/internal]: Broadcasting state change
LOG-7-BROKER: [thrd:app]: 10.60.11.3:9092/bootstrap: Added new broker with NodeId -1
LOG-7-TOPIC: [thrd:app]: New local topic: test1-b
LOG-7-BRKMAIN: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Enter main broker thread
LOG-7-TOPPARNEW: [thrd:app]: NEW test1-b [-1] 0xf85e10 (at rd_kafka_topic_new0:325)
LOG-7-CONNECT: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: broker in state INIT connecting
LOG-7-METADATA: [thrd:app]: Skipping metadata refresh of 1 topic(s): no usable brokers
LOG-7-TOPIC: [thrd:app]: New local topic: test1-u
LOG-7-TOPPARNEW: [thrd:app]: NEW test1-u [-1] 0xf82720 (at rd_kafka_topic_new0:325)
LOG-7-METADATA: [thrd:app]: Skipping metadata refresh of 1 topic(s): no usable brokers
LOG-7-CONNECT: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Connecting to ipv4#10.60.11.3:9092 (plaintext) with socket 11
LOG-7-STATE: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Broker changed state INIT -> CONNECT
LOG-7-BROADCAST: [thrd:10.60.11.3:9092/bootstrap]: Broadcasting state change
LOG-7-NOINFO: [thrd:main]: Topic test1-b partition count is zero: should refresh metadata
LOG-7-NOINFO: [thrd:main]: Topic test1-u partition count is zero: should refresh metadata
LOG-7-METADATA: [thrd:main]: Skipping metadata refresh of 2 topic(s): no usable brokers
LOG-7-NOINFO: [thrd:main]: Topic test1-b partition count is zero: should refresh metadata
LOG-7-NOINFO: [thrd:main]: Topic test1-u partition count is zero: should refresh metadata
LOG-7-METADATA: [thrd:main]: Skipping metadata refresh of 2 topic(s): no usable brokers
LOG-7-NOINFO: [thrd:main]: Topic test1-b partition count is zero: should refresh metadata
LOG-7-NOINFO: [thrd:main]: Topic test1-u partition count is zero: should refresh metadata
LOG-7-METADATA: [thrd:main]: Skipping metadata refresh of 2 topic(s): no usable brokers
LOG-7-NOINFO: [thrd:main]: Topic test1-b partition count is zero: should refresh metadata
LOG-7-NOINFO: [thrd:main]: Topic test1-u partition count is zero: should refresh metadata
LOG-7-METADATA: [thrd:main]: Skipping metadata refresh of 2 topic(s): no usable brokers
LOG-7-NOINFO: [thrd:main]: Topic test1-b partition count is zero: should refresh metadata
LOG-7-NOINFO: [thrd:main]: Topic test1-u partition count is zero: should refresh metadata
LOG-7-METADATA: [thrd:main]: Skipping metadata refresh of 2 topic(s): no usable brokers
LOG-7-NOINFO: [thrd:main]: Topic test1-b partition count is zero: should refresh metadata
LOG-7-NOINFO: [thrd:main]: Topic test1-u partition count is zero: should refresh metadata
LOG-7-METADATA: [thrd:main]: Skipping metadata refresh of 2 topic(s): no usable brokers
LOG-7-DESTROY: [thrd:app]: Terminating instance
LOG-7-TERMINATE: [thrd:app]: Interrupting timers
LOG-7-TERMINATE: [thrd:app]: Sending TERMINATE to main background thread
LOG-7-BROADCAST: [thrd:app]: Broadcasting state change
LOG-7-TERMINATE: [thrd:app]: Joining main background thread
LOG-7-DESTROY: [thrd:main]: Destroy internal
LOG-7-DESTROY: [thrd:main]: Removing all topics
LOG-7-TOPPARREMOVE: [thrd:main]: Removing toppar test1-b [-1] 0xf85e10
LOG-7-DESTROY: [thrd:main]: test1-b [-1]: 0xf85e10 DESTROY_FINAL
LOG-7-TOPPARREMOVE: [thrd:main]: Removing toppar test1-u [-1] 0xf82720
LOG-7-DESTROY: [thrd:main]: test1-u [-1]: 0xf82720 DESTROY_FINAL
LOG-7-TERMINATE: [thrd:main]: Purging reply queue
LOG-7-TERMINATE: [thrd:main]: Decommissioning internal broker
LOG-7-TERMINATE: [thrd:main]: Join 2 broker thread(s)
LOG-7-TERMINATE: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Handle is terminating: failed 0 request(s) in retry+outbuf
LOG-7-BROKERFAIL: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: failed: err: Local: Broker handle destroyed: (errno: Operation now in progress)
LOG-7-TERM: [thrd::0/internal]: :0/internal: Received TERMINATE op in state UP: 1 refcnts, 0 toppar(s), 0 active toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs
LOG-7-STATE: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Broker changed state CONNECT -> DOWN
LOG-7-BROADCAST: [thrd:10.60.11.3:9092/bootstrap]: Broadcasting state change
LOG-7-BUFQ: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Purging bufq with 0 buffers
LOG-7-BUFQ: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Updating 0 buffers on connection reset
LOG-7-TERM: [thrd:10.60.11.3:9092/bootstrap]: 10.60.11.3:9092/bootstrap: Received TERMINATE op in state DOWN: 1 refcnts, 0 toppar(s), 0 active toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs
LOG-7-TERMINATE: [thrd::0/internal]: :0/internal: Handle is terminating: failed 0 request(s) in retry+outbuf
LOG-7-BROKERFAIL: [thrd::0/internal]: :0/internal: failed: err: Local: Broker handle destroyed: (errno: Success)
LOG-7-STATE: [thrd::0/internal]: :0/internal: Broker changed state UP -> DOWN
LOG-7-BROADCAST: [thrd::0/internal]: Broadcasting state change
LOG-7-BUFQ: [thrd::0/internal]: :0/internal: Purging bufq with 0 buffers
LOG-7-BUFQ: [thrd::0/internal]: :0/internal: Updating 0 buffers on connection reset
LOG-7-TERMINATE: [thrd:main]: Main background thread exiting
LOG-7-TERMINATE: [thrd:app]: Destroying op queues
LOG-7-TERMINATE: [thrd:app]: Termination done: freeing resources

vk-coder · 2018-05-15T18:28:26Z

anymore information required? Thanks!

edenhill · 2018-05-15T18:45:45Z

It looks like that log indicates that all threads are indeed exiting.
How are you measuring threads left behind?

vk-coder · 2018-05-15T18:50:29Z

ps -eLf | grep
all reported threads are stuck in librdkafka function calls

#0 0x00007f6acc031cf2 in pthread_cond_timedwait@@GLIBC_2.3.2 () from /lib64/libpthread.so.0
#1 0x00000000004b8e65 in cnd_timedwait_ms (cnd=cnd@entry=0xab41b8, mtx=mtx@entry=0xab4190, timeout_ms=) at tinycthread.c:501
#2 0x00000000004853aa in rd_kafka_q_serve (rkq=0xab4190, timeout_ms=, max_cnt=max_cnt@entry=0, cb_type=cb_type@entry=RD_KAFKA_Q_CB_CALLBACK,
callback=callback@entry=0x0, opaque=opaque@entry=0x0) at rdkafka_queue.c:440
#3 0x000000000045a43c in rd_kafka_thread_main (arg=arg@entry=0xabf760) at rdkafka.c:1227
#4 0x00000000004b8c07 in _thrd_wrapper_function (aArg=) at tinycthread.c:624
#5 0x00007f6acc02de25 in start_thread () from /lib64/libpthread.so.0
#6 0x00007f6acaa7834d in clone () from /lib64/libc.so.6

edenhill · 2018-05-15T19:15:46Z

in gdb can you go to stack frame 3 and print p rk.rk_name to see which instance this is?
Do you see just this thread hanging or more?

vk-coder · 2018-05-16T13:03:27Z

#0 0x00007faa91a76cf2 in pthread_cond_timedwait@@GLIBC_2.3.2 () from /lib64/libpthread.so.0
Missing separate debuginfos, use: debuginfo-install glibc-2.17-196.el7_4.2.x86_64 libevent-2.0.21-4.el7.x86_64 libgcc-4.8.5-16.el7_4.1.x86_64 libstdc++-4.8.5-16.el7_4.1.x86_64 openssl-libs-1.0.2k-8.el7.x86_64 protobuf-3.3.1-2.el7.centos.x86_64 zlib-1.2.7-17.el7.x86_64
(gdb) bt
#0 0x00007faa91a76cf2 in pthread_cond_timedwait@@GLIBC_2.3.2 () from /lib64/libpthread.so.0
#1 0x00000000004bb825 in cnd_timedwait_ms (cnd=cnd@entry=0x1b2e2c8, mtx=mtx@entry=0x1b2e2a0, timeout_ms=) at tinycthread.c:501
#2 0x0000000000487a9a in rd_kafka_q_serve (rkq=0x1b2e2a0, timeout_ms=, max_cnt=max_cnt@entry=0, cb_type=cb_type@entry=RD_KAFKA_Q_CB_CALLBACK,
callback=callback@entry=0x0, opaque=opaque@entry=0x0) at rdkafka_queue.c:447
#3 0x000000000045a9bc in rd_kafka_thread_main (arg=arg@entry=0x1b39100) at rdkafka.c:1266
#4 0x00000000004bb5c7 in _thrd_wrapper_function (aArg=) at tinycthread.c:624
#5 0x00007faa91a72e25 in start_thread () from /lib64/libpthread.so.0
#6 0x00007faa904bd34d in clone () from /lib64/libc.so.6
(gdb) frame 3
#3 0x000000000045a9bc in rd_kafka_thread_main (arg=arg@entry=0x1b39100) at rdkafka.c:1266
1266 rd_kafka_q_serve(rk->rk_ops, (int)(sleeptime / 1000), 0,
(gdb) p rk.rk_name
$1 = "test1#consumer-2", '\000' <repeats 111 times>
(gdb)

edenhill · 2018-05-16T13:13:29Z

Pay attention that all outstanding librdkafka objects are destroyed prior to destroying the instance.
https://github.com/edenhill/librdkafka/wiki/Proper-termination-sequence

vk-coder · 2018-05-16T15:42:03Z

so this is what i have to do

struct rd_kafka_s *rk = _consumer->c_ptr();
delete _consumer;
rd_kafka_destroy(rk);

this stops all the threads. unfortunately this is not explicitly done in any of the examples as most of the examples call exit.

above call is for all those who want a graceful shutdown of application or shutdown and go into repeat mode.

thanks for you time.

edenhill · 2018-05-16T17:08:18Z

Calling rd_kafka_destroy(rk) should not be done from the C++ API.

vk-coder · 2018-05-18T15:54:43Z

RdKafka::KafkaConsumer was wrapped inside another container and "close()" was not getting redirected to RdKafka::KafkaConsumer::close

after fixing the this issue in my application, things are working fine as of now.

@edenhill thanks for your time and help

vk-coder closed this as completed May 16, 2018

edenhill reopened this May 16, 2018

vk-coder closed this as completed May 18, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

threads not getting cleaned up #1803

threads not getting cleaned up #1803

vk-coder commented May 11, 2018

edenhill commented May 11, 2018

vk-coder commented May 11, 2018

vk-coder commented May 15, 2018

edenhill commented May 15, 2018

vk-coder commented May 15, 2018

edenhill commented May 15, 2018

vk-coder commented May 16, 2018

edenhill commented May 16, 2018

vk-coder commented May 16, 2018

edenhill commented May 16, 2018

vk-coder commented May 18, 2018

threads not getting cleaned up #1803

threads not getting cleaned up #1803

Comments

vk-coder commented May 11, 2018

Description

How to reproduce

Checklist

edenhill commented May 11, 2018

vk-coder commented May 11, 2018

vk-coder commented May 15, 2018

edenhill commented May 15, 2018

vk-coder commented May 15, 2018

edenhill commented May 15, 2018

vk-coder commented May 16, 2018

edenhill commented May 16, 2018

vk-coder commented May 16, 2018

edenhill commented May 16, 2018

vk-coder commented May 18, 2018