rd_kafka_destroy deadlocking in consumer code

[mdorier]

I'm having an issue where rd_kafka_destroy called to destroy a consumer hangs indefinitely. When I attach gdb to the process, I find 3 threads, with the stack traces shown at the end of this post.

I couldn't reproduce this problem outside of my code base (when extracting the rdkafka calls my application makes into a single, simpler test file, I didn't get the issue) , so I suspect the issue is somewhere in my code, but I'd like to understand what happens when destroying a consumer and what could cause such a deadlock. From the stack trace, I'm assuming the main thread is trying to join the rdk:main thread, which is itself trying to join the rdk:broker1 thread, which is blocked inside rd_kafka_q_pop_serve.

For references, my code essentially does the following (producing an consuming to/from a new topic with a single partition):

• A producer rd_kafka_t and produce 100 events using a series of rd_kafka_produce_batch;
• rd_kafka_poll is called periodically;
• rd_kafka_flush is called after each call to rd_kafka_produce_batch and delivery callbacks are used to ensure all the messages in the batch have been produced;
• The producer is then destroyed.
• A consumer rd_kafka_t is created. It has enable.auto.commit set to false, auto.offset.reset set to earliest.
• rd_kafka_subscribe is used to subscribe it the topic/partition;
• rd_kafka_consumer_poll is used to poll for messages repeatedly. I can see the 100 messages coming correctly and rd_kafka_message_destroy is used on each message after the message is processed.
• I then call rd_kafka_unsubscribe on the consumer;
• Finally I call rd_kafka_destroy to destroy the consumer, this is where it blocks.

GDB stack traces:

Thread 3 (Thread 0xffff737fb020 (LWP 99059) "rdk:broker1"):
#0  __futex_abstimed_wait_common64 (private=0, cancel=true, abstime=0xffff737fa048, op=393, expected=0, futex_word=0xaaab1e3cb918) at ./nptl/futex-internal.c:57
#1  __futex_abstimed_wait_common (cancel=true, private=0, abstime=0xffff737fa048, clockid=-2013028904, expected=0, futex_word=0xaaab1e3cb918) at ./nptl/futex-internal.c:87
#2  __GI___futex_abstimed_wait_cancelable64 (futex_word=futex_word@entry=0xaaab1e3cb918, expected=expected@entry=0, clockid=clockid@entry=0, abstime=abstime@entry=0xffff737fa048, private=private@entry=0) at ./nptl/futex-internal.c:139
#3  0x0000ffff8803cc10 in __pthread_cond_wait_common (abstime=0xffff737fa048, clockid=0, mutex=0xaaab1e3cb8c0, cond=0xaaab1e3cb8f0) at ./nptl/pthread_cond_wait.c:503
#4  ___pthread_cond_timedwait64 (cond=cond@entry=0xaaab1e3cb8f0, mutex=mutex@entry=0xaaab1e3cb8c0, abstime=abstime@entry=0xffff737fa048) at ./nptl/pthread_cond_wait.c:652
#5  0x0000ffff880463ac in __cnd_timedwait (cond=cond@entry=0xaaab1e3cb8f0, mutex=mutex@entry=0xaaab1e3cb8c0, time_point=time_point@entry=0xffff737fa048) at ../sysdeps/unix/sysv/linux/cnd_timedwait.c:27
#6  0x0000ffff87df5394 in cnd_timedwait_abs (cnd=cnd@entry=0xaaab1e3cb8f0, mtx=mtx@entry=0xaaab1e3cb8c0, tspec=tspec@entry=0xffff737fa048) at tinycthread_extra.c:122
#7  0x0000ffff87d87cc4 in rd_kafka_q_pop_serve (rkq=0xaaab1e3cb8c0, timeout_us=timeout_us@entry=1000, version=version@entry=0, cb_type=cb_type@entry=RD_KAFKA_Q_CB_RETURN, callback=callback@entry=0x0, opaque=opaque@entry=0x0) at rdkafka_queue.c:473
#8  0x0000ffff87d87e50 in rd_kafka_q_pop (rkq=<optimized out>, timeout_us=timeout_us@entry=1000, version=version@entry=0) at rdkafka_queue.c:497
#9  0x0000ffff87d6cbcc in rd_kafka_broker_ops_serve (timeout_us=1000, rkb=<optimized out>) at rdkafka_broker.c:3589
#10 rd_kafka_broker_ops_io_serve (rkb=rkb@entry=0xaaab1e3b5180, abs_timeout=<optimized out>) at rdkafka_broker.c:3650
#11 0x0000ffff87d6d198 in rd_kafka_broker_consumer_serve (rkb=rkb@entry=0xaaab1e3b5180, abs_timeout=abs_timeout@entry=18115636411) at rdkafka_broker.c:4344
#12 0x0000ffff87d6d91c in rd_kafka_broker_serve (rkb=rkb@entry=0xaaab1e3b5180, timeout_ms=<optimized out>, timeout_ms@entry=1000) at rdkafka_broker.c:4486
#13 0x0000ffff87d6dd00 in rd_kafka_broker_thread_main (arg=0xaaab1e3b5180) at rdkafka_broker.c:4543
#14 0x0000ffff8803d480 in start_thread (arg=0x0) at ./nptl/pthread_create.c:439
#15 0x0000ffff880a5edc in thread_start () at ../sysdeps/unix/sysv/linux/aarch64/clone.S:79


Thread 2 (Thread 0xffff83329020 (LWP 99057) "rdk:main"):
#0  __futex_abstimed_wait_common64 (private=128, cancel=true, abstime=0x0, op=265, expected=99059, futex_word=0xffff737fb0f0) at ./nptl/futex-internal.c:57
#1  __futex_abstimed_wait_common (cancel=true, private=128, abstime=0x0, clockid=0, expected=99059, futex_word=0xffff737fb0f0) at ./nptl/futex-internal.c:87
#2  __GI___futex_abstimed_wait_cancelable64 (futex_word=futex_word@entry=0xffff737fb0f0, expected=99059, clockid=clockid@entry=0, abstime=abstime@entry=0x0, private=private@entry=128) at ./nptl/futex-internal.c:139
#3  0x0000ffff8803ef2c in __pthread_clockjoin_ex (threadid=281472619491360, thread_return=thread_return@entry=0xffff83328560, clockid=clockid@entry=0, abstime=abstime@entry=0x0, block=block@entry=true) at ./nptl/pthread_join_common.c:105
#4  0x0000ffff8803edb0 in ___pthread_join (threadid=<optimized out>, thread_return=thread_return@entry=0xffff83328560) at ./nptl/pthread_join.c:24
#5  0x0000ffff8804674c in __thrd_join (thr=<optimized out>, res=res@entry=0xffff833285bc) at ../sysdeps/pthread/thrd_join.c:26
#6  0x0000ffff87d5aac8 in rd_kafka_destroy_internal (rk=rk@entry=0xaaab1e3ccec0) at rdkafka.c:1294
#7  0x0000ffff87d5b2f0 in rd_kafka_thread_main (arg=0xaaab1e3ccec0) at rdkafka.c:2170
#8  0x0000ffff8803d480 in start_thread (arg=0x0) at ./nptl/pthread_create.c:439
#9  0x0000ffff880a5edc in thread_start () at ../sysdeps/unix/sysv/linux/aarch64/clone.S:79


Thread 1 (Thread 0xffff88701020 (LWP 99040) "KafkaEventConsu"):
#0  __futex_abstimed_wait_common64 (private=128, cancel=true, abstime=0x0, op=265, expected=99057, futex_word=0xffff833290f0) at ./nptl/futex-internal.c:57
#1  __futex_abstimed_wait_common (cancel=true, private=128, abstime=0x0, clockid=0, expected=99057, futex_word=0xffff833290f0) at ./nptl/futex-internal.c:87
#2  __GI___futex_abstimed_wait_cancelable64 (futex_word=futex_word@entry=0xffff833290f0, expected=99057, clockid=clockid@entry=0, abstime=abstime@entry=0x0, private=private@entry=128) at ./nptl/futex-internal.c:139
#3  0x0000ffff8803ef2c in __pthread_clockjoin_ex (threadid=threadid@entry=281472882872352, thread_return=thread_re--Type <RET> for more, q to quit, c to continue without paging--
turn@entry=0xfffff3e598e0, clockid=clockid@entry=0, abstime=abstime@entry=0x0, block=block@entry=true) at ./nptl/pthread_join_common.c:105
#4  0x0000ffff8803edb0 in ___pthread_join (threadid=threadid@entry=281472882872352, thread_return=thread_return@entry=0xfffff3e598e0) at ./nptl/pthread_join.c:24
#5  0x0000ffff8804674c in __thrd_join (thr=thr@entry=281472882872352, res=res@entry=0xfffff3e59954) at ../sysdeps/pthread/thrd_join.c:26
#6  0x0000ffff87d5cb6c in rd_kafka_destroy_app (rk=0xaaab1e3ccec0, flags=<optimized out>) at rdkafka.c:1131
...
#13 KafkaConsumer::~KafkaConsumer (this=0xaaab1e3cbc90, __in_chrg=<optimized out>) at /home/ubuntu/Code/kafka-test/src/KafkaConsumer.hpp:28

[mdorier]

I eventually found that the deadlock was caused by my code not calling rd_kafka_message_destroy on the very last message pulled by the consumer. Closing.