Question of rdkafka_consumer_example_cpp

[coolulu]

hi,edenhill
I use "./rdkafka_consumer_example_cpp -g group-0001 -b 192.168.18.76:9092 TEST_A" to receive
and use "./rdkafka_example_cpp -P -t TEST_A -b 192.168.18.76:9092 -p 0" to send,
but rdkafka_consumer_example_cpp not received message.
I thinks my command line wrong.
help, how to write rdkafka_consumer_example_cpp for command line?

[edenhill]

When producing the messages, do you get the same number of % Message delivery for (...): Success as messages you are producing?

[coolulu]

Message have success to delivery.

[lilu@localhost examples]$ ./rdkafka_example_cpp -P -t TEST_A -b 192.168.18.76:9092
% Created producer rdkafka#producer-1
hhhhhhhhhhhhhhhhh
% Produced message (17 bytes)
Message delivery for (17 bytes): Success <===== Successed

I have two broker, {192.168.18.76, 192.168.18.82};
The consumer print some error logs, It seems the consumers can't received any message from the other broker.

Our broker version is Kafka-0.8.2.

[lilu@localhost examples]$ ./rdkafka_consumer_example -g group-0001 -b 192.168.18.76:9092 TEST_A
1452169932.503 RDKAFKA-3-FAIL: rdkafka#consumer-1: zookeeper02:9092/2: Receive failed: Disconnected
1452169932.503 RDKAFKA-3-ERROR: rdkafka#consumer-1: zookeeper02:9092/2: Receive failed: Disconnected
1452169933.519 RDKAFKA-3-FAIL: rdkafka#consumer-1: zookeeper02:9092/2: Receive failed: Disconnected
1452169933.519 RDKAFKA-3-ERROR: rdkafka#consumer-1: zookeeper02:9092/2: Receive failed: Disconnected
1452169934.536 RDKAFKA-3-FAIL: rdkafka#consumer-1: zookeeper02:9092/2: Receive failed: Disconnected
1452169934.536 RDKAFKA-3-ERROR: rdkafka#consumer-1: zookeeper02:9092/2: Receive failed: Disconnected

[edenhill]

You will need to check your broker logs for why it is disconnecting the client.
My guess is that you might be using a 0.8 broker. The consumer groups require broker 0.9

[coolulu]

thank you, I change broker 0.9 can receive message.

rdkafka_consumer_example core dump only one time. after never appear

consumer command line is "./rdkafka_consumer_example -g GROUP -b 192.168.18.76:9092 TEST_A"
producer command line is "./rdkafka_example_cpp -P -t TEST_A -b 192.168.18.76:9092"

gdb print======================================
Reading symbols from /lib64/libnss_files-2.12.so...Reading symbols from /usr/lib/debug/lib64/libnss_files-2.12.so.debug...done.
done.
Loaded symbols for /lib64/libnss_files-2.12.so
Core was generated by `./rdkafka_consumer_example -g GROUP -b 192.168.18.76:9092 TEST_A'.
Program terminated with signal 6, Aborted.
#0 0x0000003c8d632625 in raise (sig=6) at ../nptl/sysdeps/unix/sysv/linux/raise.c:64
64 return INLINE_SYSCALL (tgkill, 3, pid, selftid, sig);
Missing separate debuginfos, use: debuginfo-install cyrus-sasl-lib-2.1.23-13.el6_3.1.x86_64 cyrus-sasl-md5-2.1.23-13.el6_3.1.x86_64 cyrus-sasl-plain-2.1.23-13.el6_3.1.x86_64 db4-4.7.25-18.el6_4.x86_64 keyutils-libs-1.4-5.el6.x86_64 krb5-libs-1.10.3-37.el6_6.x86_64 libcom_err-1.41.12-21.el6.x86_64 libselinux-2.0.94-5.8.el6.x86_64 openssl-1.0.1e-30.el6.11.x86_64
(gdb) bt
#0 0x0000003c8d632625 in raise (sig=6) at ../nptl/sysdeps/unix/sysv/linux/raise.c:64
#1 0x0000003c8d633e05 in abort () at abort.c:92
#2 0x00000000004089fb in rd_kafka_crash (file=, line=, function=, rk=0x0, reason=) at rdkafka.c:2077
#3 0x000000000041ff29 in rd_kafka_buf_callback (rkb=0x7f3d4c003c70, err=RD_KAFKA_RESP_ERR_NO_ERROR, response=0x7f3d44003c20, request=0x7f3d44002300) at rdkafka_buf.c:441
#4 0x00000000004113e7 in rd_kafka_req_response (rkb=0x7f3d4c003c70) at rdkafka_broker.c:880
#5 rd_kafka_recv (rkb=0x7f3d4c003c70) at rdkafka_broker.c:1051
#6 0x000000000041e348 in rd_kafka_transport_io_event (rktrans=0x7f3d440017e0, events=1) at rdkafka_transport.c:1008
#7 0x0000000000411c27 in rd_kafka_broker_serve (rkb=0x7f3d4c003c70, timeout_ms=0) at rdkafka_broker.c:2085
#8 0x0000000000413c4b in rd_kafka_broker_consumer_serve (arg=0x7f3d4c003c70) at rdkafka_broker.c:3191
#9 rd_kafka_broker_thread_main (arg=0x7f3d4c003c70) at rdkafka_broker.c:3279
#10 0x00000000004367ef in _thrd_wrapper_function (aArg=) at tinycthread.c:596
#11 0x0000003c8de079d1 in start_thread (arg=0x7f3d551c4700) at pthread_create.c:301
#12 0x0000003c8d6e88fd in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:115
(gdb) i threads
6 Thread 0x7f3d565c6700 (LWP 20613) 0x0000003c8d6df0d3 in __poll (fds=, nfds=, timeout=) at ../sysdeps/unix/sysv/linux/poll.c:87
5 Thread 0x7f3d55bc5700 (LWP 20614) 0x0000003c8d6df0d3 in __poll (fds=, nfds=, timeout=) at ../sysdeps/unix/sysv/linux/poll.c:87
4 Thread 0x7f3d56fc7700 (LWP 20612) pthread_cond_timedwait@@GLIBC_2.3.2 () at ../nptl/sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:239
3 Thread 0x7f3d579c8700 (LWP 20611) pthread_cond_timedwait@@GLIBC_2.3.2 () at ../nptl/sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:239
2 Thread 0x7f3d585f2360 (LWP 20610) pthread_cond_timedwait@@GLIBC_2.3.2 () at ../nptl/sysdeps/unix/sysv/linux/x86_64/pthread_cond_timedwait.S:239

• 1 Thread 0x7f3d551c4700 (LWP 20615) 0x0000003c8d632625 in raise (sig=6) at ../nptl/sysdeps/unix/sysv/linux/raise.c:64
  (gdb)

[edenhill]

Is this using the latest master?

If so, do this in gdb:

fr 3
list
p *rkbuf

[coolulu]

yes, , It is latest master, Download today
=========================
(gdb) fr 3
#3 0x000000000041ff29 in rd_kafka_buf_callback (rkb=0x7f3d4c003c70, err=RD_KAFKA_RESP_ERR_NO_ERROR, response=0x7f3d44003c20, request=0x7f3d44002300) at rdkafka_buf.c:441
441 rd_kafka_assert(NULL, !request->rkbuf_response);
(gdb) list
436 /* Decide if the request should be retried.
437 * This is always done in the originating broker thread. /
438 if (unlikely(err && rd_kafka_buf_retry(rkb, err, request)))
439 return;
440
441 rd_kafka_assert(NULL, !request->rkbuf_response);
442 request->rkbuf_response = response;
443
444 if (request->rkbuf_replyq) {
445 rd_kafka_op_t *rko = rd_kafka_op_new(RD_KAFKA_OP_RECV_BUF);
(gdb)
446
447 / Increment refcnt since rko_rkbuf will be decref:ed
448 * if q_enq() fails and we dont want the rkbuf gone in that
449 * case. /
450 rd_kafka_buf_keep(request);
451 rko->rko_rkbuf = request;
452
453 rko->rko_err = err;
454 if (rd_kafka_q_enq(request->rkbuf_replyq, rko)) {
455 / Enqueued */
(gdb) p *rkbuf
No symbol "rkbuf" in current context.
(gdb)

[edenhill]

ProTip: when pasting output, put the pasted text between two blocks of four back-ticks, .i.e.:

` `  ` `  (without spaces)
some output here

` `  ` `  (without spaces)

[coolulu]

This is good way

(gdb) fr 3
#3 0x000000000041ff29 in rd_kafka_buf_callback (rkb=0x7f3d4c003c70, err=RD_KAFKA_RESP_ERR_NO_ERROR, response=0x7f3d44003c20, request=0x7f3d44002300) at rdkafka_buf.c:441
441 rd_kafka_assert(NULL, !request->rkbuf_response);
(gdb) list
436 /* Decide if the request should be retried.
437 * This is always done in the originating broker thread. /
438 if (unlikely(err && rd_kafka_buf_retry(rkb, err, request)))
439 return;
440 
441 rd_kafka_assert(NULL, !request->rkbuf_response);
442 request->rkbuf_response = response;
443 
444 if (request->rkbuf_replyq) {
445 rd_kafka_op_t *rko = rd_kafka_op_new(RD_KAFKA_OP_RECV_BUF);
(gdb) 
446 
447 / Increment refcnt since rko_rkbuf will be decref:ed
448 * if q_enq() fails and we dont want the rkbuf gone in that
449 * case. /
450 rd_kafka_buf_keep(request);
451 rko->rko_rkbuf = request;
452 
453 rko->rko_err = err;
454 if (rd_kafka_q_enq(request->rkbuf_replyq, rko)) {
455 / Enqueued */
(gdb) p *rkbuf
No symbol "rkbuf" in current context.
(gdb)

[coolulu]

"p *rkbuf" is "p request->rkbuf_response" ?

(gdb) p request->rkbuf_response
$1 = (struct rd_kafka_buf_s *) 0x7f3d44002500
(gdb) p *request->rkbuf_response
$2 = {rkbuf_link = {tqe_next = 0x0, tqe_prev = 0x0}, rkbuf_corrid = 0, rkbuf_ts_retry = 0, rkbuf_flags = 0, rkbuf_msg = {msg_name = 0x0, msg_namelen = 0, msg_iov = 0x7f3d44002680, msg_iovlen = 1, msg_control = 0x0, 
    msg_controllen = 0, msg_flags = 0}, rkbuf_iov = 0x7f3d44002680, rkbuf_iovcnt = 1, rkbuf_of = 2, rkbuf_len = 2, rkbuf_size = 21, rkbuf_buf = 0x7f3d44002690 "", rkbuf_buf2 = 0x7f3d44001bf0 "", 
  rkbuf_rbuf = 0x7f3d44001bf0 "", rkbuf_wbuf = 0x7f3d44002690 "", rkbuf_wof = 2, rkbuf_wof_init = 21, rkbuf_crc = 0, rkbuf_reqhdr = {Size = 0, ApiKey = 0, ApiVersion = 0, CorrId = 0}, rkbuf_reshdr = {Size = 6, 
    CorrId = 475}, rkbuf_expected_size = 0, rkbuf_replyq = 0x0, rkbuf_cb = 0, rkbuf_response = 0x0, rkbuf_err = RD_KAFKA_RESP_ERR_NO_ERROR, rkbuf_rkb = 0x7f3d4c003c70, rkbuf_refcnt = {lock = {__data = {__lock = 0, 
        __count = 0, __owner = 0, __nusers = 0, __kind = 0, __spins = 0, __list = {__prev = 0x0, __next = 0x0}}, __size = '\000' <repeats 39 times>, __align = 0}, v = 1}, rkbuf_opaque = 0x0, rkbuf_op_version = 0, 
  rkbuf_retries = 0, rkbuf_ts_enq = 0, rkbuf_ts_sent = 0, rkbuf_ts_timeout = 0, rkbuf_offset = 0, rkbuf_msgq = {rkmq_msgs = {tqh_first = 0x0, tqh_last = 0x7f3d44002660}, rkmq_msg_cnt = {val = 0}, rkmq_msg_bytes = {val = 0}}}

[edenhill]

You wouldn't happen to still have that core file?
I would like a printout of p *request and p *response

[coolulu]

The core dump never seen again
A few days ago, librdkafka update after I remove core file...

Such as the problem next time, I'm telling you

[edenhill]

Okay, thanks