Librdkafka: dr_msg_cb opaque pointer is not set

[roopabayar]

Description

The opaque pointer in dr_msg_cb is not set with the value passed in rd_kafka_produce

(gdb) bt

0 dr_msg_cb (rk=0x7ffff59b6800, rkmessage=0x7ffff588b480, opaque=0x0) at ../usr/sbin/kafka-producer/kafka.c:41

1 0x0000000000426629 in rd_kafka_poll_cb (rk=0x7ffff59b6800, rkq=0x7fffffffe270, rko=0x7ffff140b240, cb_type=, opaque=) at rdkafka.c:2594

2 0x000000000044f602 in rd_kafka_q_serve (rkq=0x7ffff58af6c0, timeout_ms=, timeout_ms@entry=0, max_cnt=max_cnt@entry=0, cb_type=cb_type@entry=RD_KAFKA_Q_CB_CALLBACK,

callback=callback@entry=0x426550 <rd_kafka_poll_cb>, opaque=opaque@entry=0x0) at rdkafka_queue.c:467

3 0x000000000041d209 in rd_kafka_poll (rk=, timeout_ms=timeout_ms@entry=0) at rdkafka.c:2668

4 0x00000000004186f2 in kafka_producer_send_data (kafka_conn=0x7ffff58b6420, buf=, len=) at ../usr/sbin/kafha-producer/kafka.c:223

5 0x0000000000415619 in send_logs_to_kafka_coll (rem_coll=0x7ffff584b8c0) at

How to reproduce

Pass a valid pointer in kafka_producer_send_data
Check if the pointer is set in the callback

Checklist

Please provide the following information:

• [ x] librdkafka version (release number or git tag):

Git hash 4da1061ad1d993f446e54356b9dbc82f4ef52583
origin/0.11.0.x] [origin/HEAD] [origin/master]

• [ x] Apache Kafka version:
  0.10.2

• [ ] librdkafka client configuration:

• [ x] Operating system:
  Ubuntu linux
• [ ] 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 show the code that calls produce() and the dr_msg_cb code please?

Also note that the msg_opaque provided to produce() is rkmessage->_private in the dr_msg_cb, not the opaque argument.
https://github.com/edenhill/librdkafka/blob/master/src/rdkafka.h#L901

[roopabayar]

Hi

Below is the code snippet.

static void
dr_msg_cb (rd_kafka_t *rk,
const rd_kafka_message_t *rkmessage,
void *opaque)
{
kafka_conn_t *kafka_conn = opaque;

if (!kafka_conn) {
    return;
}

::
::
}

vs_status_t
my_kafka_producer_send_data (my_kafka_conn_t kafka_conn,
char *buf, uint16_t len)
{
if (rd_kafka_produce(kafka_conn->kafka_rkt,
RD_KAFKA_PARTITION_UA,
RD_KAFKA_MSG_F_COPY,
buf, len,
NULL, 0, kafka_conn) == -1) {
/* Failed to enqueue message for producing. */
LOG_ERROR(LOG_CONN,
"Failed to produce to topic %s: %s\n",
rd_kafka_topic_name(kafka_conn->lced_kafka_rkt),
rd_kafka_err2str(rd_kafka_last_error()));
:
:

Looks like the issue may be with rd_kafka_produce as it does not
take rd_kafka_message_t as an argument.
rd_kafka_produce_batch might be ok as i can set the _private param

thx
roopa

:

On Wed, Jun 28, 2017 at 1:37 PM, Magnus Edenhill [e-mail]
wrote:

Can you show the code that calls produce() and the dr_msg_cb code please?

Also note that the msg_opaque provided to produce() is rkmessage->_private
in the dr_msg_cb, not the opaque argument.
https://github.com/edenhill/librdkafka/blob/master/src/rdkafka.h#L901

—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub
https://github.com/edenhill/librdkafka/issues/1288#issuecomment-311781712,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AcVgbIw1kyNTKrTmqx41MirxvrxY9Ifoks5sIrl7gaJpZM4OIeau
.

[edenhill]

The msg_opaque argument to produce() will be made available as rkmessage->_private in the dr_msg_cb.
The opaque argument of dr_msg_cb is the rd_kafka_t-level global opaque.

[edenhill]

https://github.com/edenhill/librdkafka/blob/master/src/rdkafka.h#L2665

[roopabayar]

Thanks for the clarification.

Regards
roopa

On Wed, Jun 28, 2017 at 1:58 PM, Magnus Edenhill [e-mail]
wrote:

The msg_opaque argument to produce() will be made available as
rkmessage->_private in the dr_msg_cb.
The opaque argument of dr_msg_cb is the rd_kafka_t-level global opaque.

—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub
https://github.com/edenhill/librdkafka/issues/1288#issuecomment-311788585,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AcVgbEOg0SYi_DgNOJQBV5gn1ooa9nvNks5sIr5vgaJpZM4OIeau
.