Librdkafka: another possible memory leak in high-level KafkaConsumer

[alorbach]

Description

librdkafka: 0.11.x (git checkout master branch today)
kafka: 0.10.2.1 and 0.11.0
os: ubuntu 16.04 x64

Using librdkafka as consumer on a static topic. After consumed messages and shutdown, the following memory leak is detected in valgrind:
==54416== 4,039 (992 direct, 3,047 indirect) bytes in 4 blocks are definitely lost in loss record 657 of 666
==54416== at 0x4C2FB55: calloc (in /usr/lib/valgrind/vgpreload_memcheck-amd64-linux.so)
==54416== by 0x70D6073: rd_calloc (rd.h:94)
==54416== by 0x70D6073: rd_kafka_op_new0 (rdkafka_op.c:191)
==54416== by 0x70D79F9: rd_kafka_q_op_err (rdkafka_op.c:345)
==54416== by 0x70B4438: rd_kafka_fetch_reply_handle (rdkafka_broker.c:2707)
==54416== by 0x70B4E6D: rd_kafka_broker_fetch_reply (rdkafka_broker.c:2794)
==54416== by 0x70D09A2: rd_kafka_buf_callback (rdkafka_buf.c:421)
==54416== by 0x70B7224: rd_kafka_req_response (rdkafka_broker.c:1054)
==54416== by 0x70B7224: rd_kafka_recv (rdkafka_broker.c:1166)
==54416== by 0x70CE597: rd_kafka_transport_io_event (rdkafka_transport.c:1272)
==54416== by 0x70BD527: rd_kafka_broker_serve (rdkafka_broker.c:2233)
==54416== by 0x70BE3EF: rd_kafka_broker_consumer_serve (rdkafka_broker.c:3057)
==54416== by 0x70BE3EF: rd_kafka_broker_thread_main (rdkafka_broker.c:3169)
==54416== by 0x7106D56: _thrd_wrapper_function (tinycthread.c:624)
==54416== by 0x505B6B9: start_thread (pthread_create.c:333)

I am using rd_kafka_consumer_close() and rd_kafka_destroy() as described in the documentation here:
https://github.com/edenhill/librdkafka/wiki/Proper-termination-sequence

How to reproduce

I can attach debug messages from internal librdkafka starting when rd_kafka_consumer_close() is called until rd_kafka_destroy is finished.

Checklist

Please provide the following information:

• [ ] librdkafka version (release number or git tag):
• [ ] 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

[alorbach]

kafka-consumer-log.txt

[edenhill]

Are you returning all rd_kafka_message_t objects to librdkafka by calling rd_kafka_message_destroy()?

Is this reproducible with one of the tools in examples/?

[alorbach]

Parts of my code is based on "examples/rdkafka_consumer_example.c".
I am using rd_kafka_message_destroy to free rkmessage after I have processed it.

I have not tried to reproduce the issue using the example code.

[ltagliamonte]

@edenhill any updates on this memory leak issues? they are blocking.

[alorbach]

I am sorry I caused confusion. I didn't see that we have supression for this memory leak.
@ltagliamonte It only concerns the kafka consumer.

The leak still exists.

[edenhill]

Can you provide a small reproducible program?

[ChenyuanHu]

I have also encountered this problem.

I am commenting on the following function, no longer leaked.

rd_kafka_fetch_reply_handle
....
2692 case RD_KAFKA_RESP_ERR_MSG_SIZE_TOO_LARGE:
2693 default: /* and all other errors */
2694 rd_dassert(tver->version > 0);
2695 #if 0 // comment this and no longer leaked
2696 rd_kafka_q_op_err(
2697 rktp->rktp_fetchq,
2698 RD_KAFKA_OP_CONSUMER_ERR,
2699 hdr.ErrorCode, tver->version,
2700 rktp,
2701 rktp->rktp_offsets.fetch_offset,
2702 "%s",
2703 rd_kafka_err2str(hdr.ErrorCode));
2704 #endif
2705 printf("hh kafka debug %d, %sn", hdr.ErrorCode, rd_kafka_err2str(hdr.ErrorCode));
2706 break;
2707 }

After do this，I got the log below and no longer leaked.

hh kafka debug -191, Broker: No more messages
hh kafka debug -191, Broker: No more messages
hh kafka debug -191, Broker: No more messages
hh kafka debug -191, Broker: No more messages
hh kafka debug -191, Broker: No more messages
hh kafka debug -191, Broker: No more messages
hh kafka debug -191, Broker: No more messages

==========================================================

I check my comsumer code and find the bug final. I am not called rd_kafka_message_destroy when rkmsg->err != RD_KAFKA_RESP_ERR_NO_ERROR.

It is not a bug.

[alorbach]

Damn you are right @ChenyuanHu thanks for the hint.
That's exactly what happened in my case.

I feel stupid now... ;)

[edenhill]

Glad to hear you found it :)

[rgerhards]

@edenhill thanks for your support!

[yijia2413]

@ChenyuanHu :+1: