consumer crash

[yanzhiqiang]

Description

consumer crash ,
(gdb) bt
#0 0x00007f446b2ed625 in raise () from /lib64/libc.so.6
#1 0x00007f446b2eee05 in abort () from /lib64/libc.so.6
#2 0x00007f44606dbb5b in rd_kafka_crash (file=, line=, function=, rk=0x7f444402e2d0, reason=) at rdkafka.c:2571
#3 0x00007f44606f8cc9 in rd_kafka_offset_broker_commit (rktp=0x7f444402f3f0) at rdkafka_offset.c:562
#4 rd_kafka_offset_commit (rktp=0x7f444402f3f0) at rdkafka_offset.c:618
#5 0x00007f44606f8e28 in rd_kafka_offset_store_stop (rktp=0x7f444402f3f0) at rdkafka_offset.c:1008
#6 0x00007f4460713765 in rd_kafka_toppar_fetch_stop (rktp=0x7f444402f3f0, rko_orig=0x7f444404a570) at rdkafka_partition.c:1330
#7 0x00007f44606da549 in rd_kafka_toppar_q_cb (rk=, rko=, cb_type=, opaque=) at rdkafka.c:1024
#8 0x00007f446070076d in rd_kafka_q_serve (rkq=0x7f44440118b0, timeout_ms=, max_cnt=, cb_type=0, callback=0x7f44606da540 <rd_kafka_toppar_q_cb>, opaque=0x0)

at rdkafka_queue.c:422
#9 0x00007f44606e0b64 in rd_kafka_toppars_q_serve (arg=0x7f444402e2d0) at rdkafka.c:1030
#10 rd_kafka_thread_main (arg=0x7f444402e2d0) at rdkafka.c:1070
#11 0x00007f446071ce5f in _thrd_wrapper_function (aArg=) at tinycthread.c:613
#12 0x00007f446a7ddaa1 in start_thread () from /lib64/libpthread.so.0
#13 0x00007f446b3a393d in clone () from /lib64/libc.so.6

How to reproduce

1：create producer
2：producer messages
3：create some consumers
4：consume all messages

Checklist

Please provide the following information:

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

[edenhill]

Can you post your consumer shutdown code? (e.g., when you stop consuming and start tearing it down)

[edenhill]

Also provide your consumer config.

[yanzhiqiang]

my shutdown code is
i have two threads,
one thread code is:
rd_kafka_poll(rk, 10);
other thread stop code is：
1： rd_kafka_consume_stop(rkt, partition);
2：wait poll thread done
3：while (rd_kafka_outq_len(rk) > 0)
rd_kafka_poll(rk, 10);
4：rd_kafka_topic_destroy(rkt);
5：rd_kafka_destroy(rk);
my consumer config is:
auto.commit.enable true
auto.commit.interval.ms 5000

[edenhill]

Did you configure a group.id? That's required for broker based offset commits to work.
Also, what broker version are you on?

[yanzhiqiang]

i have set group.id, broker version is 0.9.0.0

[edenhill]

Okay, can you reproduce with debug=cgrp,topic enabled and upload the logs?

[edenhill]

You should also consider using the high-level consumer (subscribe() or assign() instead)