Which commit fixed Python confluent-kafka version 1.8.2 process hang at cnd_timedwait () -> futex call

[ns-btao]

Initially, we noticed a confluent-kafka python hang issue after a power outage of kafka brokers.

We reproduced this issue with the following steps:

1. Start the process to consume messages from kafka brokers, at this moment, everything is working.
2. Find out all kafka brokers IPs, all DNS IPs.
3. Applied iptable rules to drop all the packages between them to simulate disconnections.
4. Leave the process in the disconnected state for a long time(I wait for more than one day, I don't know if there is any specific time we should wait).
5. Flush the iptable rules, and all the sudden, the process hangs.

Observations:
there are 12 threads running, and mainly can divided to doing two things, one part of threads are waiting for, with RD_POLL_INFINITE, and I don't see any code may signal the rkq_cond to wake up these threads. I think this is why it looks like a hang:

cnd_timedwait_abs(&rkq->rkq_cond, &rkq->rkq_lock, &timeout_tspec)

second part of the threads are polling events:

rd_kafka_transport_poll(...)

And from our process log, during the disconnection:

%3|1663369211.564|FAIL|consumer-10743#consumer-1| [thrd:GroupCoordinator]: GroupCoordinator: kafka01:9092: Failed to resolve 'kafka01:9092': Temporary failure in name resolution (after 0ms in state CONNECT, 4 identical error(s) suppressed)
%3|1663369211.877|FAIL|monitor#consumer-3| [thrd:kafka01:9092/bootstrap]: kafka01:9092/1: Failed to resolve 'kafka01:9092': Temporary failure in name resolution (after 0ms in state CONNECT, 25 identical error(s) suppressed)

And after flush iptable rules, it looks like a hang, and no more logs.

GDB backtrace information:

(gdb) info thread
  Id   Target Id         Frame 
* 1    Thread 0x7f0f715bd700 (LWP 8144) "python" pthread_cond_wait@@GLIBC_2.3.2
    () at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_wait.S:185
  2    Thread 0x7f0f69488700 (LWP 8252) "rdk:broker-1" 0x00007f0f7036584d in poll () at ../sysdeps/unix/syscall-template.S:84
  3    Thread 0x7f0f68c87700 (LWP 8253) "rdk:main" pthread_cond_timedwait@@GLIBC_2.3.2 () at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
  4    Thread 0x7f0f63fff700 (LWP 8254) "rdk:broker-1" pthread_cond_timedwait@@GLIBC_2.3.2 () at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
  5    Thread 0x7f0f637fe700 (LWP 8255) "rdk:broker1" 0x00007f0f7036584d in poll () at ../sysdeps/unix/syscall-template.S:84
  6    Thread 0x7f0f62ffd700 (LWP 8256) "rdk:main" pthread_cond_timedwait@@GLIBC_2.3.2 () at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
  7    Thread 0x7f0f627fc700 (LWP 8257) "rdk:broker-1" pthread_cond_timedwait@@GLIBC_2.3.2 () at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
  8    Thread 0x7f0f61ffb700 (LWP 8258) "rdk:broker1" 0x00007f0f7036584d in poll () at ../sysdeps/unix/syscall-template.S:84
  9    Thread 0x7f0f617fa700 (LWP 9684) "rdk:broker-1" 0x00007f0f7036584d in poll () at ../sysdeps/unix/syscall-template.S:84
  10   Thread 0x7f0f60ff9700 (LWP 9685) "rdk:main" pthread_cond_timedwait@@GLIBC_2.3.2 () at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
  11   Thread 0x7f0f47fff700 (LWP 9686) "rdk:broker-1" pthread_cond_timedwait@@GLIBC_2.3.2 () at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
  12   Thread 0x7f0f477fe700 (LWP 9687) "rdk:broker1" pthread_cond_timedwait@@GLIBC_2.3.2 () at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
  (gdb) thread 1
[Switching to thread 1 (Thread 0x7f0f715bd700 (LWP 8144))]
#0  pthread_cond_wait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_wait.S:185
185	../sysdeps/unix/sysv/linux/x86_64/pthread_cond_wait.S: No such file or directory.
(gdb) bt
#0  pthread_cond_wait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_wait.S:185
#1  0x00007f0f69781d59 in cnd_wait ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#2  0x00007f0f6972cd5f in rd_kafka_q_pop_serve.localalias ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#3  0x00007f0f69783809 in rd_kafka_metadata ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#4  0x00007f0f69e20e7c in list_topics (self=0x7f0f6afbb680, 
    args=<optimized out>, kwargs=<optimized out>)
    at /tmp/pip-req-build-l8c5hs_4/src/confluent_kafka/src/confluent_kafka.h:448
(gdb) thread 2
[Switching to thread 2 (Thread 0x7f0f69488700 (LWP 8252))]
#0  0x00007f0f7036584d in poll () at ../sysdeps/unix/syscall-template.S:84
84	../sysdeps/unix/syscall-template.S: No such file or directory.
(gdb) bt
#0  0x00007f0f7036584d in poll () at ../sysdeps/unix/syscall-template.S:84
#1  0x00007f0f697278f9 in rd_kafka_transport_poll ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#2  0x00007f0f69727974 in rd_kafka_transport_io_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#3  0x00007f0f6970f825 in rd_kafka_broker_ops_io_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#4  0x00007f0f69710098 in rd_kafka_broker_consumer_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#5  0x00007f0f6971165b in rd_kafka_broker_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#6  0x00007f0f69711d1d in rd_kafka_broker_thread_main ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#7  0x00007f0f69781be7 in _thrd_wrapper_function ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#8  0x00007f0f7118e6ba in start_thread (arg=0x7f0f69488700)
    at pthread_create.c:333
#9  0x00007f0f7037151d in clone ()
    at ../sysdeps/unix/sysv/linux/x86_64/clone.S:109

(gdb) thread 3
[Switching to thread 3 (Thread 0x7f0f68c87700 (LWP 8253))]
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
225	../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S: No such file or directory.
(gdb) bt
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
#1  0x00007f0f69781d79 in cnd_timedwait ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#2  0x00007f0f6972cf8f in rd_kafka_q_serve.localalias ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#3  0x00007f0f696f6f04 in rd_kafka_thread_main ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#4  0x00007f0f69781be7 in _thrd_wrapper_function ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#5  0x00007f0f7118e6ba in start_thread (arg=0x7f0f68c87700)
    at pthread_create.c:333
#6  0x00007f0f7037151d in clone ()
    at ../sysdeps/unix/sysv/linux/x86_64/clone.S:109
(gdb) thread 4
[Switching to thread 4 (Thread 0x7f0f63fff700 (LWP 8254))]
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
225	in ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S
(gdb) bt
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
#1  0x00007f0f69781d79 in cnd_timedwait ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#2  0x00007f0f6972cd5f in rd_kafka_q_pop_serve.localalias ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#3  0x00007f0f6970f73f in rd_kafka_broker_ops_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#4  0x00007f0f6970f87a in rd_kafka_broker_ops_io_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#5  0x00007f0f6971198f in rd_kafka_broker_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#6  0x00007f0f69711eb9 in rd_kafka_broker_thread_main ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#7  0x00007f0f69781be7 in _thrd_wrapper_function ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#8  0x00007f0f7118e6ba in start_thread (arg=0x7f0f63fff700)
    at pthread_create.c:333
#9  0x00007f0f7037151d in clone ()
    at ../sysdeps/unix/sysv/linux/x86_64/clone.S:109
(gdb) thread 6
[Switching to thread 6 (Thread 0x7f0f62ffd700 (LWP 8256))]
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
225	../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S: No such file or directory.
(gdb) bt
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
#1  0x00007f0f69781d79 in cnd_timedwait ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#2  0x00007f0f6972cf8f in rd_kafka_q_serve.localalias ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#3  0x00007f0f696f6f04 in rd_kafka_thread_main ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#4  0x00007f0f69781be7 in _thrd_wrapper_function ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#5  0x00007f0f7118e6ba in start_thread (arg=0x7f0f62ffd700)
    at pthread_create.c:333
#6  0x00007f0f7037151d in clone ()
    at ../sysdeps/unix/sysv/linux/x86_64/clone.S:109
(gdb) thread 7
[Switching to thread 7 (Thread 0x7f0f627fc700 (LWP 8257))]
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
225	in ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S
(gdb) bt
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
#1  0x00007f0f69781d79 in cnd_timedwait ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#2  0x00007f0f6972cd5f in rd_kafka_q_pop_serve.localalias ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#3  0x00007f0f6970f73f in rd_kafka_broker_ops_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#4  0x00007f0f6970f87a in rd_kafka_broker_ops_io_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#5  0x00007f0f69711829 in rd_kafka_broker_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#6  0x00007f0f69711eb9 in rd_kafka_broker_thread_main ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#7  0x00007f0f69781be7 in _thrd_wrapper_function ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#8  0x00007f0f7118e6ba in start_thread (arg=0x7f0f627fc700)
    at pthread_create.c:333
#9  0x00007f0f7037151d in clone ()
    at ../sysdeps/unix/sysv/linux/x86_64/clone.S:109
(gdb) thread 10
[Switching to thread 10 (Thread 0x7f0f60ff9700 (LWP 9685))]
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
225	in ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S
(gdb) bt
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
#1  0x00007f0f69781d79 in cnd_timedwait ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#2  0x00007f0f6972cf8f in rd_kafka_q_serve.localalias ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#3  0x00007f0f696f6f04 in rd_kafka_thread_main ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#4  0x00007f0f69781be7 in _thrd_wrapper_function ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#5  0x00007f0f7118e6ba in start_thread (arg=0x7f0f60ff9700)
    at pthread_create.c:333
#6  0x00007f0f7037151d in clone ()
    at ../sysdeps/unix/sysv/linux/x86_64/clone.S:109
(gdb) thread 11
[Switching to thread 11 (Thread 0x7f0f47fff700 (LWP 9686))]
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
225	in ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S
(gdb) bt
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
#1  0x00007f0f69781d79 in cnd_timedwait ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#2  0x00007f0f6972cd5f in rd_kafka_q_pop_serve.localalias ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#3  0x00007f0f6970f73f in rd_kafka_broker_ops_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#4  0x00007f0f6970f87a in rd_kafka_broker_ops_io_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#5  0x00007f0f6971198f in rd_kafka_broker_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#6  0x00007f0f69711eb9 in rd_kafka_broker_thread_main ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#7  0x00007f0f69781be7 in _thrd_wrapper_function ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#8  0x00007f0f7118e6ba in start_thread (arg=0x7f0f47fff700)
    at pthread_create.c:333
#9  0x00007f0f7037151d in clone ()
    at ../sysdeps/unix/sysv/linux/x86_64/clone.S:109
(gdb) thread 12
[Switching to thread 12 (Thread 0x7f0f477fe700 (LWP 9687))]
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
225	in ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S
(gdb) bt
#0  pthread_cond_timedwait@@GLIBC_2.3.2 ()
    at ../sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:225
#1  0x00007f0f69781d79 in cnd_timedwait ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#2  0x00007f0f6972cd5f in rd_kafka_q_pop_serve.localalias ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#3  0x00007f0f6970f73f in rd_kafka_broker_ops_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#4  0x00007f0f6970f87a in rd_kafka_broker_ops_io_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#5  0x00007f0f69710098 in rd_kafka_broker_consumer_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#6  0x00007f0f6971165b in rd_kafka_broker_serve ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#7  0x00007f0f69711eb9 in rd_kafka_broker_thread_main ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#8  0x00007f0f69781be7 in _thrd_wrapper_function ()
   from /lib/python3.8/site-packages/confluent_kafka/../confluent_kafka.libs/librdkafka-ba99fd30.so.1
#9  0x00007f0f7118e6ba in start_thread (arg=0x7f0f477fe700)
    at pthread_create.c:333
#10 0x00007f0f7037151d in clone ()
    at ../sysdeps/unix/sysv/linux/x86_64/clone.S:109

Please help me to understand this problem, and point me which commit fixes this issue. Thanks.

[ns-btao]

Can anyone help me to confirm this commit: coord_req: re-query coordinator to avoid getting stuck fixes this issue? Here is my understanding, please correct me if I am wrong:

During the disconnection, we see this log repeatly:

%3|1663374676.274|FAIL|producer-29534#producer-1| [thrd:kafka01-auto-dc1:9092/bootstrap]: kafka01-auto-dc1:9092/1: Failed to resolve 'kafka01-auto-dc1:9092': Temporary failure in name resolution (after 0ms in state CONNECT, 15 identical error(s) suppressed)

And based on the code:

        if (!rkb->rkb_rsal) {
                        rd_kafka_broker_fail(rkb, LOG_ERR,                      
                                             RD_KAFKA_RESP_ERR__RESOLVE,
                                             "Failed to resolve '%s': %s",
                                             nodename, errstr);

And in function rd_kafka_broker_fail, it will set the state to DOWN:

    /* Set broker state */
        old_state = rkb->rkb_state;
    rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_DOWN);

And in the 1.8.2 librdkafka function rd_kafka_broker_set_state, it missed this condition, which has been fixed in the above commit:

                /* If the connection or connection attempt failed and there
                 * are coord_reqs or cgrp awaiting this coordinator to come up
                 * then trigger the monitors so that rd_kafka_coord_req_fsm()
                 * is triggered, which in turn may trigger a new coordinator
                 * query. */

And because of this fix, it will trigger the monitor ops:

        if (trigger_monitors)
                rd_kafka_broker_trigger_monitors(rkb);

And inside function rd_kafka_broker_trigger_monitors, it will call rd_kafka_q_enq(rkbmon->rkbmon_q, rko);, and in the code flows, it will call cnd_signal(&rkq->rkq_cond); to wakeup the threads which are waiting for this condition variable, like here, where our threads are waiting for in function rd_kafka_q_pop_serve:

                        if (cnd_timedwait_abs(&rkq->rkq_cond,                   
                                              &rkq->rkq_lock,
                                              &timeout_tspec) !=
                            thrd_success) { 

And this is how this issue gets fixed.
Please correct me if my understanding is wrong. Thanks.

[edenhill]

That fix is only relevant for consumers and transactional producers since it relies on the persistent-connection coordinator counter, which is not used by the non-txn producer, unless there's an admin API call being made (e.g., CreateTopics()).

[ns-btao]

@edenhill , makes sense to me. So my understand of the code flow looks correct to you?