Core rd_kafka_buf_grow assertion failure during rapid usage of consumer

[trthulhu]

Description

We are running on a 3-node distributed system that attempts to poll data from a 3-node Kafka cluster using rdkafka. We have 20 topics (single partitions) we are polling every 2 seconds from 10 threads across the cluster (ie 1 node will probably poll from 3/4 topics at a time, with 6-8 total per 2 second window).

One node will core and bring down the distributed system after some hours (ranging so far from 6-15 hours).

Each 2 second cycle includes a complete process of: creating rdkafka conf, handle, queue, then polling data from a topic for a couple hundred milliseconds before closing/destroying everything. 3/4 of these cycles will be happening at a time in parallel (different topics) per node.

Here is the backtrace from the aborting thread:
#0 0x00007fac178f4989 in raise () from /lib64/libc.so.6
#1 0x00007fac178f6098 in abort () from /lib64/libc.so.6
#2 0x00007fabe5054a03 in rd_kafka_crash (file=, line=,

function=<optimized out>, rk=0x0, reason=<optimized out>) at rdkafka.c:2478

#3 0x00007fabe5073bf2 in rd_kafka_buf_grow (rkbuf=0x7f9c3c9a0060, needed_len=)

at rdkafka_buf.c:145

#4 0x00007fabe50851e0 in rd_kafka_buf_write (len=9, data=0x7fa9af180960, rkbuf=0x7f9c3c9a0060)

at rdkafka_buf.h:374

#5 rd_kafka_buf_write_kstr (kstr=0x7fa9af180950, rkbuf=0x7f9c3c9a0060) at rdkafka_buf.h:512
#6 rd_kafka_MetadataRequest0 (rkb=0x7fa9afa54270, all_topics=, only_rkt=0x0,

reason=<optimized out>) at rdkafka_request.c:1511

#7 0x00007fabe505f3ab in rd_kafka_broker_metadata_req_op (rkb=rkb@entry=0x7fa9afa54270,

rko=rko@entry=0x7fa9ae014f10) at rdkafka_broker.c:791

#8 0x00007fabe5066d0b in rd_kafka_broker_op_serve (rkb=rkb@entry=0x7fa9afa54270, rko=0x7fa9ae014f10)

at rdkafka_broker.c:2855

#9 0x00007fabe506765b in rd_kafka_broker_serve (rkb=rkb@entry=0x7fa9afa54270,

timeout_ms=timeout_ms@entry=10) at rdkafka_broker.c:3099

#10 0x00007fabe50679e1 in rd_kafka_broker_ua_idle (rkb=rkb@entry=0x7fa9afa54270,

timeout_ms=timeout_ms@entry=0) at rdkafka_broker.c:3160

#11 0x00007fabe5068227 in rd_kafka_broker_thread_main (arg=arg@entry=0x7fa9afa54270)

at rdkafka_broker.c:4487

#12 0x00007fabe5098877 in _thrd_wrapper_function (aArg=) at tinycthread.c:613
#13 0x00007fac174a6df3 in start_thread () from /lib64/libpthread.so.0
#14 0x00007fac179b53dd in clone () from /lib64/libc.so.6

The log file has this message:

*** rdkafka_buf.c:145:rd_kafka_buf_grow: assert: rkbuf->rkbuf_flags & RD_KAFKA_OP_F_FREE ***

which can be found here: rdkafka-log.txt

Note: this log complains a lot about an inability to connect to localhost, which is interesting because we do not give localhost as a parameter to the broker list ever. Right before the core we see request timeout: disconnects from 2 kafka nodes, however the brokers never went down (afaik).

How to reproduce

Difficult without a long running test with our system. The reproducer I created for this issue attempts to replicate the process, but we have not used it to replicate this particular issue.

Checklist

Please provide the following information:

• librdkafka version (release number or git tag): master @ 0d83cff
• Apache Kafka version: 0.9
• librdkafka client configuration:
• Operating system: Redhat 7
• Using the legacy Consumer
• Using the high-level KafkaConsumer
• Provide logs (with debug=.. as necessary) from librdkafka
• Provide broker log excerpts
• Critical issue

[edenhill]

Re the localhost connection attempts: maybe the broker reports such endpoints in the metadata reply?
You can verify that by doing [rdkafka_example|kafkacat] -b <somebroker> -L

[edenhill]

Irrelevant to the problem but Im curious:
Why are you having such short lived consumers that only run for two seconds?
Did you see the pause() API?
https://github.com/edenhill/librdkafka/blob/master/src/rdkafka.h#L1501

[trthulhu]

Our API is SQL, and we utilize rdkafka via a User-Defined function within a SQL COPY statement (ex: COPY foo KafkaSource(parameters... like broker list, kafka topic/partition/offsets, duration of time to run, etc) ). When someone runs this statement in SQL we do a full cycle of creating/polling/tearing down rdkafka handles.

An application on top of this uses these statements for micro-batching data transactionally into the database (along with offsets) atomically. The duration each statement lasts is configurable, but customers like as close to real-time as possible, hence: 2 second use cases.

Ideally, we should expect this use case and keep the handles and queues alive between API calls, but currently we do not (as this would require a deeper integration with out main product outside the scope of the SDK we use for User-Defined functionality). Hope that helps!

Also re: localhost -- just checked with kafkacat and received the real endpoints in the metadata request.

[edenhill]

Thanks for the explanation, your current use makes sense.

Re localhost, right, it must come from bootstrap.servers property or brokers_add() API then.
librdkafka is not making it up on its own :)

[edenhill]

Re the real issue here:

I've seen a case where this happens is when rd_kafka_topic_conf_t objects are reused for multiple topics.
Can you verify that you do not reuse the conifg object passed to topic_new()? If you want to reuse it you must make a copy (topic_conf_dup()).

[trthulhu]

I can confirm we do not share the rd_kafka_topic_conf_t with more than a single topic.

[edenhill]

Can you recompile librdkafka by using ./dev-conf.sh and reproduce the issue.
Then pop up gdb and do:

bt full
fr 3 (or wherever grow is)
p *rkbuf
fr 6
p *rkt

Thanks

[trthulhu]

Sure thing. Reproduce may take awhile, I'll let you know when it occurs, thanks!

[edenhill]

Do you create/destroy, or start/stop topics during the process lifetime more than at the beginning?

[trthulhu]

Hmm... I believe each time a SQL command is executed (that runs rdkafka), we do the whole cycle: create/run/destroy. We only create at the beginning and destroy at the end of this cycle. However, the process in question is the main process of our database system (different threads). So... yes.

[edenhill]

And you are careful with only using rd_kafka_topic_t objects with the rd_kafka_t handle they were created by?

[trthulhu]

Yes, I'm pretty sure that each cycle only opens one rd_kafka_t handle, then create all the topic_t handles from this. Then, at the end, all topic_t handles are closed and immediately following the kafka_t handle is closed.

These could be happening in parallel, of course. So two rd_kafka_t handles could be open to the same cluster and even be consuming from the same topic-partition (though usually not the same partition).

They would each have their own set of handles though.

[edenhill]

Do you want me to review the relevant parts of the code?

[trthulhu]

Possibly? I'll let you know after discussing with people here.

[trthulhu]

I think we are trying to set up something formal for the above, fyi.

[edenhill]

Have you seen this again?

[trthulhu]

Not yet :( (we haven't tried really, been concentrating on other issues) -- got limited resources at the moment so we're trying to prioritize timing of things. Really sorry about that, will let you know though, I did not forget.

[nagaprabhu]

Hi, I just ran into the same issue and able to reproduce it with ./rdkafka_example_cpp. The same works fine when I exclude compression.
rdkafka_example_cpp.txt

[edenhill]

@nagaprabhu Thank you, I can now reproduce it the same. I think the underlying issue is for messages that grow after compression.

[edenhill]

Actually the issue reported by @nagaprabhu is different (Producer compression code) than what is reported orginally by @panarchus in this issue (Metadata request code).

[trthulhu]

We're reproducing (trying) now... so hopefully have some more details over weekend or Monday. Also, we have a theory:

the crash appears to happen when writing all the topic names to a buffer in order to generate a metadata request and the assert fails when the buffer is too small and non-growable. The buffer isn't growable because rdkafka counts the number of topics and creates a fixed-size buffer from that count. Is it possible a race condition could occur if topics are being added during this counting?

[edenhill]

Thanks for your troubleshooting efforts!

Your idea has bearing, but both of the iterations are protected by the same lock:
https://github.com/edenhill/librdkafka/blob/master/src/rdkafka_request.c#L1490

And all inserts and removals are also protected by that same rk_lock.

[arnaud-lb]

I have a similar issue, here is some code reproducing it 100% of the times: https://gist.github.com/arnaud-lb/b62c60c5dbd3a0e69ff7407d66ad63a4

Output:

*** rdkafka_buf.c:149:rd_kafka_buf_grow: assert: rkbuf->rkbuf_flags & RD_KAFKA_OP_F_FREE ***

Backtrace:

#0  0x00007ffff77d1067 in __GI_raise (sig=sig@entry=6) at ../nptl/sysdeps/unix/sysv/linux/raise.c:56
#1  0x00007ffff77d2448 in __GI_abort () at abort.c:89
#2  0x00007ffff7b59906 in rd_kafka_crash (file=0x7ffff7bbba62 "rdkafka_buf.c", line=149, function=0x7ffff7bbbdb0 <__FUNCTION__.19818> "rd_kafka_buf_grow",
    rk=0x0, reason=0x7ffff7bbbab8 "assert: rkbuf->rkbuf_flags & RD_KAFKA_OP_F_FREE") at rdkafka.c:2481
#3  0x00007ffff7b80ddc in rd_kafka_buf_grow (rkbuf=0x7fffd8000cf0, needed_len=220) at rdkafka_buf.c:149
#4  0x00007ffff7b804da in rd_kafka_buf_write (rkbuf=0x7fffd8000cf0, data=0x7fffd8042730, len=131) at rdkafka_buf.h:382
#5  0x00007ffff7b818a3 in rd_kafka_buf_write_Message (rkb=0x7fffe8002d50, rkbuf=0x7fffd8000cf0, Offset=0, MagicByte=0 '\000', Attributes=1 '\001',
    Timestamp=0, key=0x0, key_len=0, payload=0x7fffd8042730, len=131, outlenp=0x7fffe7ffc1d4) at rdkafka_buf.c:413
#6  0x00007ffff7b6702c in rd_kafka_compress_MessageSet_buf (rkb=0x7fffe8002d50, rktp=0x7fffe8009380, rkbuf=0x7fffd8000cf0, iov_firstmsg=2, of_firstmsg=63,
    of_init_firstmsg=63, MsgVersion=0, timestamp_firstmsg=0, MessageSetSizep=0x7fffe7ffc2bc) at rdkafka_broker.c:2648
#7  0x00007ffff7b67690 in rd_kafka_broker_produce_toppar (rkb=0x7fffe8002d50, rktp=0x7fffe8009380) at rdkafka_broker.c:2821
#8  0x00007ffff7b68db9 in rd_kafka_toppar_producer_serve (rkb=0x7fffe8002d50, rktp=0x7fffe8009380, do_timeout_scan=1, now=1037244134017)
    at rdkafka_broker.c:3243
#9  0x00007ffff7b68f4f in rd_kafka_broker_producer_serve (rkb=0x7fffe8002d50) at rdkafka_broker.c:3299
#10 0x00007ffff7b703b9 in rd_kafka_broker_thread_main (arg=0x7fffe8002d50) at rdkafka_broker.c:4514
#11 0x00007ffff7badcf9 in _thrd_wrapper_function (aArg=0x7fffe8003480) at tinycthread.c:613
#12 0x00007ffff75870a4 in start_thread (arg=0x7fffe7fff700) at pthread_create.c:309
#13 0x00007ffff788462d in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:111

librdkafka: master b484fc0

[arnaud-lb]

I realise that this may be the same issue than @nagaprabhu's

[edenhill]

@arnaud-lb Yep, looks to be the same, messages growing after compression, which has been fixed on the partition_changes branch.
Big kudos for the reproducible test case! 💯

[arnaud-lb]

Is it safe to cherry pick 45b730a on master ?

[edenhill]

I believe so

Den 5 okt. 2016 4:58 em skrev "Arnaud Le Blanc" [email protected]:

Is it safe to cherry pick 45b730a
45b730a
on master ?

—
You are receiving this because you commented.
Reply to this email directly, view it on GitHub
#781 (comment),
or mute the thread
https://github.com/notifications/unsubscribe-auth/AAgCvlMM6NvHqqSUzMa1_Zi1vzDLo6rQks5qw7sIgaJpZM4J3R0H
.

[edenhill]

@panarchus Have you seen this again?

[trthulhu]

We updated our library to a more recent one and have done repeated trials and have not been able to reproduce. Note: not the most recent master branch, but one from 2-3 weeks ago. This could be good news. If you want, you can close this and in the off-chance we find it again I'll reopen.

[edenhill]

Okay, sounds good.
Thanks for your effort!