Librdkafka: memory leaks after shutdown (with compression.codec=snappy)

[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 producer with "compression.codec" option other than none will create memory leaks findable with valgrind.

How to reproduce

Set compression.codec=snappy for example.
Run kafka producer application with valgrind and terminate.

Checklist

Please provide the following information:

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

Make sure to run valgrind with the suppressions file: valgrind --suppressions=librdkafka/tests/librdkafka.suppressions ./your-program ...

If you can reproduce the leak with the suppression file then please provide the output from valgrind.

[edenhill]

I can't reproduce this with rdkafka_performance ... -z snappy

[alorbach]

I am going to add librdkafka.supressions into my tests to see if the result changes.
I had memory leaks with all compression options (snappy, gzip and lz4).

[alorbach]

The supression file does not change anything.

[edenhill]

Please provide the valgrind leak report.

Also please show me your produce() call and dr_msg_cb.

[alorbach]

Here are the valgrind reports when using snappy:

==92769== 92 bytes in 1 blocks are possibly lost in loss record 564 of 657
==92769== at 0x4C2DB8F: malloc (in /usr/lib/valgrind/vgpreload_memcheck-amd64-linux.so)
==92769== by 0x71157A8: rd_malloc (rd.h:100)
==92769== by 0x71157A8: rd_kafka_msgset_writer_compress_snappy.isra.12 (rdkafka_msgset_writer.c:835)
==92769== by 0x711739F: rd_kafka_msgset_writer_compress (rdkafka_msgset_writer.c:915)
==92769== by 0x711739F: rd_kafka_msgset_writer_finalize (rdkafka_msgset_writer.c:1113)
==92769== by 0x711739F: rd_kafka_msgset_create_ProduceRequest (rdkafka_msgset_writer.c:1160)
==92769== by 0x70EB605: rd_kafka_ProduceRequest (rdkafka_request.c:1823)
==92769== by 0x70C3664: rd_kafka_toppar_producer_serve (rdkafka_broker.c:2373)
==92769== by 0x70C3664: rd_kafka_broker_producer_serve (rdkafka_broker.c:2432)
==92769== by 0x70C3664: rd_kafka_broker_thread_main (rdkafka_broker.c:3167)
==92769== by 0x710BD56: _thrd_wrapper_function (tinycthread.c:624)
==92769== by 0x505B6B9: start_thread (pthread_create.c:333)
==92769== by 0x5D983DC: clone (clone.S:109)
==92769==
==92769== 10,639 bytes in 57 blocks are definitely lost in loss record 656 of 657
==92769== at 0x4C2DB8F: malloc (in /usr/lib/valgrind/vgpreload_memcheck-amd64-linux.so)
==92769== by 0x71157A8: rd_malloc (rd.h:100)
==92769== by 0x71157A8: rd_kafka_msgset_writer_compress_snappy.isra.12 (rdkafka_msgset_writer.c:835)
==92769== by 0x711739F: rd_kafka_msgset_writer_compress (rdkafka_msgset_writer.c:915)
==92769== by 0x711739F: rd_kafka_msgset_writer_finalize (rdkafka_msgset_writer.c:1113)
==92769== by 0x711739F: rd_kafka_msgset_create_ProduceRequest (rdkafka_msgset_writer.c:1160)
==92769== by 0x70EB605: rd_kafka_ProduceRequest (rdkafka_request.c:1823)
==92769== by 0x70C3664: rd_kafka_toppar_producer_serve (rdkafka_broker.c:2373)
==92769== by 0x70C3664: rd_kafka_broker_producer_serve (rdkafka_broker.c:2432)
==92769== by 0x70C3664: rd_kafka_broker_thread_main (rdkafka_broker.c:3167)
==92769== by 0x710BD56: _thrd_wrapper_function (tinycthread.c:624)
==92769== by 0x505B6B9: start_thread (pthread_create.c:333)
==92769== by 0x5D983DC: clone (clone.S:109)

[alorbach]

producev call is here:
https://github.com/alorbach/rsyslog-1/blob/kafka-termination-problem/plugins/omkafka/omkafka.c#L611

Delivery Callback is here: https://github.com/alorbach/rsyslog-1/blob/kafka-termination-problem/plugins/omkafka/omkafka.c#L704

[edenhill]

Can you verify that all messages are completely flushed before destroying the client, specifically this debug message should be seen:
https://github.com/alorbach/rsyslog-1/blob/kafka-termination-problem/plugins/omkafka/omkafka.c#L776

[alorbach]

Yes there are some messages flushed:
5297.726687941:main thread : omkafka.c: omkafka: onDestroyflushed remaining '2' messages to kafka topic 'static'
5297.726980472:main thread : omkafka.c: omkafka: onDestroy kafka outqueue length: 0, callbacks called 0
5297.727501476:main thread : omkafka.c: omkafka: closing topic static

This happens after rd_kafka_destroy call:
5297.728570641:main thread : omkafka.c: kafka log message [7,DESTROY]: [thrd:app]: Terminating instance
5297.728887328:main thread : omkafka.c: kafka log message [7,TERMINATE]: [thrd:app]: Interrupting timers
5297.729550825:main thread : omkafka.c: kafka log message [7,TERMINATE]: [thrd:app]: Sending TERMINATE to main background thread
5297.730396884:main thread : omkafka.c: kafka log message [7,BROADCAST]: [thrd:app]: Broadcasting state change
5297.730736159:main thread : omkafka.c: kafka log message [7,TERMINATE]: [thrd:app]: Joining main background thread
5297.733121312:bc3d700: omkafka.c: kafka log message [7,DESTROY]: [thrd:main]: Destroy internal
5297.733901850:bc3d700: omkafka.c: kafka log message [7,DESTROY]: [thrd:main]: Removing all topics
5297.738221808:bc3d700: omkafka.c: kafka log message [7,PARTCNT]: [thrd:main]: Topic static partition count changed from 2 to 0
5297.738609926:bc3d700: omkafka.c: kafka log message [7,REMOVE]: [thrd:main]: static [0] no longer reported in metadata
5297.739599815:bc3d700: omkafka.c: kafka log message [7,BRKMIGR]: [thrd:main]: static [0] 0x61f5b20 sending final LEAVE for removal by ubuntu16.fritz.box:29092/0
5297.740993010:bc3d700: omkafka.c: kafka log message [7,REMOVE]: [thrd:main]: static [1] no longer reported in metadata
5297.741099243:bc3d700: omkafka.c: kafka log message [7,BRKMIGR]: [thrd:main]: static [1] 0x61f6050 sending final LEAVE for removal by ubuntu16.fritz.box:29092/0
5297.741938013:bc3d700: omkafka.c: kafka log message [7,TOPPARREMOVE]: [thrd:main]: Removing toppar static [-1] 0x61efe50
5297.742807229:bc3d700: omkafka.c: kafka log message [7,DESTROY]: [thrd:main]: static [-1]: 0x61efe50 DESTROY_FINAL
5297.746424407:bc3d700: omkafka.c: kafka log message [7,TERMINATE]: [thrd:main]: Purging reply queue
5297.747045324:bc3d700: omkafka.c: kafka log message [7,TERMINATE]: [thrd:main]: Decommissioning internal broker
5297.748555930:bc3d700: omkafka.c: kafka log message [7,TERMINATE]: [thrd:main]: Join 3 broker thread(s)
5297.748934936:d440700: omkafka.c: kafka log message [7,TOPPAR]: [thrd:ubuntu16.fritz.box:29092/0]: ubuntu16.fritz.box:29092/0: static [1] 0+0 msgs
5297.749019426:d440700: omkafka.c: kafka log message [7,TOPPAR]: [thrd:ubuntu16.fritz.box:29092/0]: ubuntu16.fritz.box:29092/0: static [0] 0+0 msgs
5297.750365645:d440700: omkafka.c: kafka log message [7,TOPBRK]: [thrd:ubuntu16.fritz.box:29092/0]: ubuntu16.fritz.box:29092/0: Topic static [0]: leaving broker (0 messages in xmitq, next leader (none), rktp 0x61f5b20)
5297.752047547:d440700: omkafka.c: kafka log message [7,TOPBRK]: [thrd:ubuntu16.fritz.box:29092/0]: ubuntu16.fritz.box:29092/0: Topic static [0]: no next leader, failing 0 message(s) in partition queue
5297.752877593:d440700: omkafka.c: kafka log message [7,BROADCAST]: [thrd:ubuntu16.fritz.box:29092/0]: Broadcasting state change
5297.753261949:d440700: omkafka.c: kafka log message [7,TOPPARREMOVE]: [thrd:ubuntu16.fritz.box:29092/0]: Removing toppar static [0] 0x61f5b20
5297.753356430:d440700: omkafka.c: kafka log message [7,DESTROY]: [thrd:ubuntu16.fritz.box:29092/0]: static [0]: 0x61f5b20 DESTROY_FINAL
5297.754713830:d440700: omkafka.c: kafka log message [7,TOPBRK]: [thrd:ubuntu16.fritz.box:29092/0]: ubuntu16.fritz.box:29092/0: Topic static [1]: leaving broker (0 messages in xmitq, next leader (none), rktp 0x61f6050)
5297.754827373:d440700: omkafka.c: kafka log message [7,TOPBRK]: [thrd:ubuntu16.fritz.box:29092/0]: ubuntu16.fritz.box:29092/0: Topic static [1]: no next leader, failing 0 message(s) in partition queue
5297.754919858:d440700: omkafka.c: kafka log message [7,BROADCAST]: [thrd:ubuntu16.fritz.box:29092/0]: Broadcasting state change
5297.755014035:d440700: omkafka.c: kafka log message [7,TOPPARREMOVE]: [thrd:ubuntu16.fritz.box:29092/0]: Removing toppar static [1] 0x61f6050
5297.755108753:d440700: omkafka.c: kafka log message [7,DESTROY]: [thrd:ubuntu16.fritz.box:29092/0]: static [1]: 0x61f6050 DESTROY_FINAL
5297.757107880:d440700: omkafka.c: kafka log message [7,TERM]: [thrd:ubuntu16.fritz.box:29092/0]: ubuntu16.fritz.box:29092/0: Received TERMINATE op in state UP: 1 refcnts, 0 toppar(s), 0 fetch toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs
5297.757346854:c43e700: omkafka.c: kafka log message [7,TERM]: [thrd::0/internal]: :0/internal: Received TERMINATE op in state UP: 1 refcnts, 0 toppar(s), 0 fetch toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs
5297.851998596:cc3f700: omkafka.c: kafka log message [7,TERMINATE]: [thrd:localhost:29092/bootstrap]: localhost:29092/bootstrap: Handle is terminating: failed 0 request(s) in retry+outbuf
5297.856910039:cc3f700: omkafka.c: kafka log message [7,BROKERFAIL]: [thrd:localhost:29092/bootstrap]: localhost:29092/bootstrap: failed: err: Local: Broker handle destroyed: (errno: Resource temporarily unavailable)
5297.861723325:cc3f700: omkafka.c: kafka log message [7,STATE]: [thrd:localhost:29092/bootstrap]: localhost:29092/bootstrap: Broker changed state UP -> DOWN
5297.863288186:cc3f700: omkafka.c: kafka log message [7,BROADCAST]: [thrd:localhost:29092/bootstrap]: Broadcasting state change
5297.867067299:cc3f700: omkafka.c: kafka log message [7,BUFQ]: [thrd:localhost:29092/bootstrap]: localhost:29092/bootstrap: Purging bufq with 0 buffers
5297.869222627:cc3f700: omkafka.c: kafka log message [7,BUFQ]: [thrd:localhost:29092/bootstrap]: localhost:29092/bootstrap: Updating 0 buffers on connection reset
5297.878604767:d440700: omkafka.c: kafka log message [7,TERMINATE]: [thrd:ubuntu16.fritz.box:29092/0]: ubuntu16.fritz.box:29092/0: Handle is terminating: failed 0 request(s) in retry+outbuf
5297.879858078:d440700: omkafka.c: kafka log message [7,BROKERFAIL]: [thrd:ubuntu16.fritz.box:29092/0]: ubuntu16.fritz.box:29092/0: failed: err: Local: Broker handle destroyed: (errno: Resource temporarily unavailable)
5297.880143946:d440700: omkafka.c: kafka log message [7,STATE]: [thrd:ubuntu16.fritz.box:29092/0]: ubuntu16.fritz.box:29092/0: Broker changed state UP -> DOWN
5297.880320430:d440700: omkafka.c: kafka log message [7,BROADCAST]: [thrd:ubuntu16.fritz.box:29092/0]: Broadcasting state change
5297.880558888:d440700: omkafka.c: kafka log message [7,BUFQ]: [thrd:ubuntu16.fritz.box:29092/0]: ubuntu16.fritz.box:29092/0: Purging bufq with 0 buffers
5297.880744555:d440700: omkafka.c: kafka log message [7,BUFQ]: [thrd:ubuntu16.fritz.box:29092/0]: ubuntu16.fritz.box:29092/0: Updating 0 buffers on connection reset
5297.881112727:c43e700: omkafka.c: kafka log message [7,TERMINATE]: [thrd::0/internal]: :0/internal: Handle is terminating: failed 0 request(s) in retry+outbuf
5297.881330963:c43e700: omkafka.c: kafka log message [7,BROKERFAIL]: [thrd::0/internal]: :0/internal: failed: err: Local: Broker handle destroyed: (errno: Success)
5297.881831800:c43e700: omkafka.c: kafka log message [7,STATE]: [thrd::0/internal]: :0/internal: Broker changed state UP -> DOWN
5297.882024180:c43e700: omkafka.c: kafka log message [7,BROADCAST]: [thrd::0/internal]: Broadcasting state change
5297.882207468:c43e700: omkafka.c: kafka log message [7,BUFQ]: [thrd::0/internal]: :0/internal: Purging bufq with 0 buffers
5297.882384884:c43e700: omkafka.c: kafka log message [7,BUFQ]: [thrd::0/internal]: :0/internal: Updating 0 buffers on connection reset
5297.888428681:bc3d700: omkafka.c: kafka log message [7,TERMINATE]: [thrd:main]: Main background thread exiting
5297.895090325:main thread : omkafka.c: kafka log message [7,TERMINATE]: [thrd:app]: Destroying op queues
5297.897134366:main thread : omkafka.c: kafka log message [7,TERMINATE]: [thrd:app]: Termination done: freeing resources

[edenhill]

Thank you, I've now reproduced it locally.

[ltagliamonte]

@edenhill thank you for the wonderful library, this is issue is a blocker for me, do you have any update?

[rgerhards]

@edenhill do you have some insight of what triggers this problem? I think librdkafka is fairly well used. So I would expect that all others using compression (aka "everyone") has this problem. This doesn't seem to be the case. So what are we doing that triggers it? Can we work around the bug via some way?

Any feedback would be deeply appreciated.

[edenhill]

I'm not able to reproduce it anymore, so it seems to be environment and/or timing related.

If you can provide a small reproducible program, or using one of the librdkafka/examples (such as rdkafka_performance -z ..) that'd be great.

Thanks

[rgerhards]

@alorbach is this one now also solved (as an rsyslog bug)? Or is it sill open?

[alorbach]

Yes the leak is still present in omkafka.

[edenhill]

This is now fixed on master. Thanks for the reports!

[rgerhards]

@edenhill thanks for fixing it albeit we had no clear repro. Much appreciated (speaking for rsyslog users here as well). We'll update soon and re-run our test bench with compression enabled.