Librdkafka: core related to broker refcnt and DNS issues

[trthulhu]

Description

We are running into an interesting issue using rdkafka that results in a core file with this backtrace:

0 0x00007f5f4b58bc10 in pthread_mutex_lock () from /lib64/libpthread.so.0

1 0x00007f5f3036cc49 in mtx_lock (mtx=) at tinycthread.c:135

2 0x00007f5f3033b386 in rd_refcnt_sub0 (R=0x1e0) at rd.h:266

3 rd_kafka_broker_op_serve (rko=0x7f5c2c02b690, rkb=0x7f5c100d27b0) at rdkafka_broker.c:2027

4 rd_kafka_broker_serve (rkb=rkb@entry=0x7f5c100d27b0, timeout_ms=timeout_ms@entry=10) at rdkafka_broker.c:2072

5 0x00007f5f3033cb91 in rd_kafka_broker_ua_idle (rkb=rkb@entry=0x7f5c100d27b0, timeout_ms=timeout_ms@entry=1000) at rdkafka_broker.c:2136

6 0x00007f5f3033cd98 in rd_kafka_broker_thread_main (arg=0x7f5c100d27b0) at rdkafka_broker.c:3277

7 0x00007f5f3036cb99 in _thrd_wrapper_function (aArg=) at tinycthread.c:596

8 0x00007f5f4b589df3 in start_thread () from /lib64/libpthread.so.0

9 0x00007f5f4ba983dd in clone () from /lib64/libc.so.

We run into this problem on our distributed database (Vertica, 3 physical nodes, each loading and using librdkafka) on nodes in which the Kafka brokers/zookeeper do not have DNS names in /etc/hosts, and we give the IP addresses to rdkafka. It will usually work for about 30 minutes or so before crashing. If we include the DNS names in all /etc/hosts files the crash does not occur.

The database attempts to load data from topics every 10 seconds, and each invocation we re-connect/give rdkafka the broker list.

Have you seen any similar issues? Check below for some environment details. Let me know what other information would be useful for you. Also, is this expected?

Thanks!

Checklist

Please provide the following information:

• [X] librdkafka version (release number or git tag): 0.9.1
• [X] Apache Kafka version: kafka_2.11-0.9.0.1
• [ ] librdkafka client configuration:
• [X] Operating system: CentOS 7
• [X] Using the legacy Consumer
• [ ] Provide logs (with debug=.. as necessary) from librdkafka
• [ ] Provide broker log excerpts
• [X] Critical issue

[edenhill]

Hey,

havent seen that before.

Can you open up gdb with that core and do:

fr 3
list
# try to make out what object it is calling refcnt_sub on and then do: p *obj

[trthulhu]

I will do this when I load the symbols; in the meantime @ frame 4:
(gdb) p *rkb
$4 = {rkb_link = {tqe_next = 0x7f5c100d2e10, tqe_prev = 0x7f5c100a8620}, rkb_nodeid = 1, rkb_rsal = 0x0,
rkb_t_rsal_last = 0, rkb_addr_last = 0x0, rkb_transport = 0x0, rkb_corrid = 2, rkb_ops = {rkq_lock = {
__data = {__lock = 0, __count = 0, __owner = 0, __nusers = 0, __kind = 0, __spins = 0, __list = {
__prev = 0x0, __next = 0x0}}, __size = '\000' , __align = 0}, rkq_cond = {
__data = {__lock = 0, __futex = 202, __total_seq = 101, __wakeup_seq = 101, __woken_seq = 101,
__mutex = 0x7f5c100d27f0, __nwaiters = 0, __broadcast_seq = 0},
__size = "\000\000\000\000\312\000\000\000e\000\000\000\000\000\000\000e\000\000\000\000\000\000\000e\000\000\000\000\000\000\000\360'\r\020\177\000\000\000\000\000\000\000\000\000", __align = 867583393792},
rkq_fwdq = 0x0, rkq_q = {tqh_first = 0x0, tqh_last = 0x7f5c100d2850}, rkq_qlen = 0, rkq_qsize = 0,
rkq_refcnt = 1, rkq_flags = 2, rkq_rk = 0x7f5b700fe850}, rkb_toppars = {tqh_first = 0x0,
tqh_last = 0x7f5c100d2880}, rkb_toppar_cnt = -1, rkb_fetch_toppars = {cqh_first = 0x7f5c100d2898,
cqh_last = 0x7f5c100d2898}, rkb_fetch_toppar_cnt = 0, rkb_fetch_toppar_next = 0x0, rkb_cgrp = 0x0,
rkb_ts_fetch_backoff = 0, rkb_fetching = 0, rkb_state = RD_KAFKA_BROKER_STATE_DOWN,
rkb_ts_state = 3383218937259, rkb_timeout_scan_intvl = {ri_ts_last = 3383218947452, ri_fixed = 0,
ri_backoff = 0}, rkb_blocking_request_cnt = {val = 0}, rkb_source = RD_KAFKA_LEARNED, rkb_c = {
tx_bytes = {val = 0}, tx = {val = 0}, tx_err = {val = 0}, tx_retries = {val = 0}, req_timeouts = {
val = 0}, rx_bytes = {val = 0}, rx = {val = 0}, rx_err = {val = 0}, rx_corrid_err = {val = 0},
rx_partial = {val = 0}}, rkb_req_timeouts = 0, rkb_ts_metadata_poll = 3383519913812,
rkb_metadata_fast_poll_cnt = 0, rkb_lock = {__data = {__lock = 0, __count = 0, __owner = 0, __nusers = 0,
__kind = 0, __spins = 0, __list = {__prev = 0x0, __next = 0x0}}, __size = '\000' ,
__align = 0}, rkb_thread = 140033717696256, rkb_refcnt = {lock = {__data = {__lock = 0, __count = 0,
__owner = 0, __nusers = 0, __kind = 0, __spins = 0, __list = {__prev = 0x0, __next = 0x0}},
__size = '\000' , __align = 0}, v = 2}, rkb_rk = 0x7f5b700fe850, rkb_recv_buf = 0x0,
rkb_outbufs = {rkbq_bufs = {tqh_first = 0x7f5c100a7500, tqh_last = 0x7f5c100a7500}, rkbq_cnt = {val = 1},
rkbq_msg_cnt = {val = 0}}, rkb_outbuf_msgcnt = {val = 0}, rkb_waitresps = {rkbq_bufs = {tqh_first = 0x0,
tqh_last = 0x7f5c100d29f0}, rkbq_cnt = {val = 0}, rkbq_msg_cnt = {val = 0}}, rkb_retrybufs = {
rkbq_bufs = {tqh_first = 0x0, tqh_last = 0x7f5c100d2a08}, rkbq_cnt = {val = 0}, rkbq_msg_cnt = {
val = 0}}, rkb_avg_rtt = {ra_v = {maxv = 0, minv = 0, avg = 0, sum = 0, cnt = 0,
start = 3383218936812}, ra_lock = {__data = {__lock = 0, __count = 0, __owner = 0, __nusers = 0,
__kind = 0, __spins = 0, __list = {__prev = 0x0, __next = 0x0}}, __size = '\000' ,
__align = 0}, ra_type = RD_AVG_GAUGE}, rkb_avg_throttle = {ra_v = {maxv = 0, minv = 0, avg = 0,
sum = 0, cnt = 0, start = 3383218936812}, ra_lock = {__data = {__lock = 0, __count = 0, __owner = 0,
__nusers = 0, __kind = 0, __spins = 0, __list = {__prev = 0x0, __next = 0x0}},
---Type to continue, or q to quit---
__size = '\000' , __align = 0}, ra_type = RD_AVG_GAUGE},
rkb_name = "intco64-181:9093/1", '\000' ,
rkb_nodename = "intco64-181:9093", '\000' , rkb_port = 9093,
rkb_origname = 0x7f5c10001290 "intco64-181", rkb_ts_connect = 3383219939992,
rkb_proto = RD_KAFKA_PROTO_PLAINTEXT, rkb_err = {
msg = "intco64-181:9093/1: Failed to resolve 'intco64-181:9093': Name or service not known", '\000' , err = 14}}

That msg value shows the DNS name for the IP address of the brokers... could this be related to: https://issues.apache.org/jira/browse/KAFKA-3352? Also, I'm pretty sure the broker didn't go down.

I will have the info you asked for later today.

[trthulhu]

fr 3, line 2027: rd_kafka_broker_destroy(rktp->rktp_leader);
(gdb) p rktp->rktp_leader
$5 = (rd_kafka_broker_t *) 0x0
(gdb) p *(rktp->rktp_leader)
Cannot access memory at address 0x0

(gdb) p rktp
$11 = (rd_kafka_toppar_t *) 0x7f5b700b86a0
(gdb) p *rktp
$12 = {rktp_removed = 0, rktp_rklink = {tqe_next = 0x0, tqe_prev = 0x0}, rktp_rkblink = {tqe_next = 0x0,
tqe_prev = 0x7f5c100d2880}, rktp_fetchlink = {cqe_next = 0x0, cqe_prev = 0x0}, rktp_rktlink = {
tqe_next = 0x0, tqe_prev = 0x0}, rktp_cgrplink = {tqe_next = 0x0, tqe_prev = 0x0},
rktp_rkt = 0x7f5b70103690, rktp_s_rkt = 0x7f5b70103690, rktp_partition = 0, rktp_leader = 0x0,
rktp_next_leader = 0x7f5b700122b0, rktp_refcnt = {lock = {__data = {__lock = 0, __count = 0, __owner = 0,
__nusers = 0, __kind = 0, __spins = 0, __list = {__prev = 0x0, __next = 0x0}},
__size = '\000' , __align = 0}, v = 3}, rktp_lock = {__data = {__lock = 2,
__count = 0, __owner = 343350, __nusers = 1, __kind = 0, __spins = 0, __list = {__prev = 0x0,
__next = 0x0}},
__size = "\002\000\000\000\000\000\000\000\066=\005\000\001", '\000' , __align = 2},
rktp_msgq = {rkmq_msgs = {tqh_first = 0x0, tqh_last = 0x7f5b700b8778}, rkmq_msg_cnt = {val = 0},
rkmq_msg_bytes = {val = 0}}, rktp_xmit_msgq = {rkmq_msgs = {tqh_first = 0x0, tqh_last = 0x7f5b700b8798},
rkmq_msg_cnt = {val = 0}, rkmq_msg_bytes = {val = 0}}, rktp_ts_last_xmit = 0, rktp_fetch = 0,
rktp_fetchq = {rkq_lock = {__data = {__lock = 0, __count = 0, __owner = 0, __nusers = 0, __kind = 0,
__spins = 0, __list = {__prev = 0x0, __next = 0x0}}, __size = '\000' ,
__align = 0}, rkq_cond = {__data = {__lock = 0, __futex = 0, __total_seq = 0, __wakeup_seq = 0,
__woken_seq = 0, __mutex = 0x0, __nwaiters = 0, __broadcast_seq = 0},
__size = '\000' , __align = 0}, rkq_fwdq = 0x0, rkq_q = {tqh_first = 0x0,
tqh_last = 0x7f5b700b8828}, rkq_qlen = 0, rkq_qsize = 0, rkq_refcnt = 1, rkq_flags = 0,
rkq_rk = 0x7f5b700fe850}, rktp_ops = {rkq_lock = {__data = {__lock = 0, __count = 0, __owner = 0,
__nusers = 0, __kind = 0, __spins = 0, __list = {__prev = 0x0, __next = 0x0}},
__size = '\000' , __align = 0}, rkq_cond = {__data = {__lock = 0, __futex = 0,
__total_seq = 0, __wakeup_seq = 0, __woken_seq = 0, __mutex = 0x0, __nwaiters = 0,
__broadcast_seq = 0}, __size = '\000' , __align = 0}, rkq_fwdq = 0x7f5b700fe8e0,
rkq_q = {tqh_first = 0x0, tqh_last = 0x7f5b700b88b8}, rkq_qlen = 0, rkq_qsize = 0, rkq_refcnt = 2,
rkq_flags = 0, rkq_rk = 0x7f5b700fe850}, rktp_version = {val = 3}, rktp_fetch_version = 0,
rktp_fetch_state = RD_KAFKA_TOPPAR_FETCH_OFFSET_WAIT, rktp_query_offset = -1, rktp_next_offset = 0,
rktp_app_offset = 0, rktp_stored_offset = -1, rktp_committing_offset = 0, rktp_committed_offset = -1,
rktp_ts_committed_offset = 0, rktp_offsets = {fetch_offset = 0, eof_offset = -1, hi_offset = -1},
rktp_offsets_fin = {fetch_offset = 0, eof_offset = -1, hi_offset = -1}, rktp_lo_offset = -1,
rktp_ts_offset_lag = 0, rktp_offset_path = 0x0, rktp_offset_fp = 0x0, rktp_cgrp = 0x0, rktp_assigned = 0,
rktp_replyq = 0x0, rktp_flags = 0, rktp_s_for_desp = 0x0, rktp_s_for_cgrp = 0x0, rktp_s_for_rkb = 0x0,
rktp_offset_query_tmr = {rtmr_link = {tqe_next = 0x7f5c82ffc0f0, tqe_prev = 0x7f5b700fec80},
rtmr_next = 0, rtmr_interval = 0, rtmr_callback = 0x7f5f303480a0 ,
rtmr_arg = 0x7f5b700b86a0}, rktp_offset_commit_tmr = {rtmr_link = {tqe_next = 0x0, tqe_prev = 0x0},
rtmr_next = 0, rtmr_interval = 0, rtmr_callback = 0x0, rtmr_arg = 0x0}, rktp_offset_sync_tmr = {
rtmr_link = {tqe_next = 0x0, tqe_prev = 0x0}, rtmr_next = 0, rtmr_interval = 0, rtmr_callback = 0x0,
rtmr_arg = 0x0}, rktp_consumer_lag_tmr = {rtmr_link = {tqe_next = 0x0, tqe_prev = 0x0}, rtmr_next = 0,
rtmr_interval = 0, rtmr_callback = 0x0, rtmr_arg = 0x0}, rktp_c = {tx_msgs = {val = 0}, tx_bytes = {
val = 0}, msgs = {val = 0}, rx_ver_drops = {val = 0}}}

[edenhill]

Any chance you still have that core file and could type 'list' at that frame? Those line numbers dont really match up with mine..

https://github.com/edenhill/librdkafka/blob/0.9.1/src/rdkafka_broker.c#L2027

[trthulhu]

Here's the source:

           /* Remove from fetcher list */
    rd_kafka_toppar_fetch_decide(rktp, rkb, 1/*force remove*/);

            rd_kafka_toppar_lock(rktp);

            rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_TOPIC, "TOPBRK",
                       "Topic %s [%"PRId32"]: leaving broker "
                       "(next leader %s)",
                       rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition,
                       rktp->rktp_next_leader ?
                       rktp->rktp_next_leader->rkb_name : "(none)");

    TAILQ_REMOVE(&rkb->rkb_toppars, rktp, rktp_rkblink);
    rkb->rkb_toppar_cnt--;
            rd_kafka_broker_destroy(rktp->rktp_leader); //---------------> LINE 2027
    rktp->rktp_leader = NULL;

            /* Need to hold on to a refcount past q_enq() and
             * unlock() below */
            s_rktp = rktp->rktp_s_for_rkb;
            rktp->rktp_s_for_rkb = NULL;

            if (rktp->rktp_next_leader) {
                    /* There is a next leader we need to migrate to. */
                    rko->rko_type = RD_KAFKA_OP_PARTITION_JOIN;
                    rd_kafka_q_enq(&rktp->rktp_next_leader->rkb_ops, rko);
                    rko = NULL;
            }

[edenhill]

Thanks!

I have an idea what the problem might be, a race condition that takes place during a quick succession of partition leader changes, but it seems a bit far fetched.

If this is somewhat easily reproducible I would like you to reproduce with debug=broker,topic enabled

[trthulhu]

rdkafka configuration?

[edenhill]

yes sir

[trthulhu]

I will see what I can do :)

[trthulhu]

Haven't been able to reproduce it quite yet... it seems the environment that the team was working on is in use. I'll let you know more when it's available.

[trthulhu]

Alright, sorry for the long wait! Finally got a reproducer. Attaching a log file.
rdkafka-dns-log.txt

[edenhill]

Great! Do you have a corresponding core file which you can provide the above info from?
Thanks

[trthulhu]

Unfortunately no... I haven't been able to reproduce myself and am relying on a strange environment from another. Sorry for being a bad reporter! I'll talk with them again to see if we can get a core and corresponding debug output.

[edenhill]

Reopen if reseen