Program does not terminate when producer and consumer are in the same program

[jbimbert]

Hello,

running a producer thread and a consumer thread in the same kafka manager, I got stuck when terminating the program. Both threads exit successfully, but the main thread does not exit. Running with debug=metadata,broker,topic,protocol I have the following logs that repeats endlessly:

%7|1671033809.365|RECV|rdkafka#producer-1| [thrd:bktv2800:9092/bootstrap]: bktv2800:9092/0: Received MetadataResponse (v2, 401 bytes, CorrId 62, rtt 27.47ms)
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: bktv2800:9092/0: ===== Received metadata (for 1 requested topics): periodic refresh =====
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: bktv2800:9092/0: ClusterId: onBXn9UrSzCGI7uUagYd_Q, ControllerId: 1
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: bktv2800:9092/0: 6 brokers, 1 topics
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: bktv2800:9092/0: Broker #0/6: bktv2800:9092 NodeId 0
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: bktv2800:9092/0: Broker #1/6: bktv2805:9092 NodeId 5
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: bktv2800:9092/0: Broker #2/6: bktv2804:9092 NodeId 4
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: bktv2800:9092/0: Broker #3/6: bktv2801:9092 NodeId 1
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: bktv2800:9092/0: Broker #4/6: bktv2802:9092 NodeId 2
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: bktv2800:9092/0: Broker #5/6: bktv2803:9092 NodeId 3
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: bktv2800:9092/0: Topic #0/1: jbi-test with 3 partitions
%7|1671033809.365|TERMINATE|rdkafka#consumer-2| [thrd:bktv2800:9092/bootstrap]: bktv2800:9092/0: Handle is terminating in state INIT: 3 refcnts (0x712690), 0 toppar(s), 0 active toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs: failed 0 request(s) in retry+outbuf
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: Topic jbi-test partition 0 Leader 0
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: Topic jbi-test partition 1 Leader 5
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: Topic jbi-test partition 2 Leader 4
%7|1671033809.365|METADATA|rdkafka#producer-1| [thrd:main]: bktv2800:9092/0: 1/1 requested topic(s) seen in metadata
%7|1671033809.365|TERMINATE|rdkafka#consumer-2| [thrd:bktv2804:9092/4]: bktv2804:9092/4: Handle is terminating in state INIT: 2 refcnts (0x7f11e0003450), 0 toppar(s), 0 active toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs: failed 0 request(s) in retry+outbuf
%7|1671033809.365|TERMINATE|rdkafka#consumer-2| [thrd:bktv2805:9092/5]: bktv2805:9092/5: Handle is terminating in state INIT: 3 refcnts (0x7f11e0002660), 0 toppar(s), 0 active toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs: failed 0 request(s) in retry+outbuf

Here I have a cluster of 6 brokers bktv2800..bktv2805
One topic : jbi-test with 3 partitions (p0->leader0, p1->leader5, p2->leader4)
One consumer and one producer

Synopsis of the program:

• create the config
• create the consumer
• create the producer
• run a thread with the consumer that consumes 5 messages and return
• run a thread with the producer that produces 5 messages and return
• wait for the consumer thread join
• wait for producer thread join
• return

consumer uses_kafkaHandlerC = rd_kafka_new(RD_KAFKA_CONSUMER, iRdConf...) and I call rd_kafka_consumer_close(_kafkaHandlerC)

producer uses _kafkaHandlerP = rd_kafka_new(RD_KAFKA_PRODUCER, iRdConf...) and I call rd_kafka_destroy(_kafkaHandlerP)

I destroy as well the topic with rd_kafka_topic_destroy(...)

Is it something wrong to use both consumer and producer in the same program?
Should I clean the consumer/producer differently?

Best regards,
JB

[edenhill]

https://github.com/edenhill/librdkafka/blob/master/INTRODUCTION.md#termination

[jbimbert]

Hello Endenhill, thank you for the quick answer.
Yes I have checked that first, and the prog runs under valgrind, etc. to be sure of no leak.

more tests:
Test1:

• loop 2 times { std::thread(consume, ...); join(); } => no issue (messages produced with a kafka-console-producer.sh elsewhere are read correctly)
• loop 2 times { std::thread(consume, ...); produce(); join(); } => stuck when returning from main()!

Test2:

• loop 2 times { std::thread(produce, ...) } => no issue (messages are produced and I get the messages in a kafka-console-consumer.sh elsewhere)
• loop 2 times { std::thread(produce, ...); consume(); join(); } => stuck when returning from main()!

Test3:

• { std::thread(produce, ...); std::thread(consume, ...); join(); join(); } => no issue (callbacks show the messages perfectly)
• { produce(); consume(); } => no issue (idem)

1. It happens only when I loop and either the consumer or the producer is in a thread... But I have all the messages produced and consumed successfully in all tests (the callbacks run perfectly for both consumer and producer).
2. Maybe something worth noting : sometimes it works fine... but most of the times, an run into the issue
3. Whatever the threads sequence, when exiting from main(), I notice that the program hangs in rd_kafka_destroy() of the consumer, always. I never managed to get the rd_kafka_destroy() of the producer first...

I know from the logs that there is 2 or 3 refcnts still not freed, but I can't know which ones...

Do you have a hint as how to see the recfcnts target?

[jbimbert]

You were right :)
During the consumer poll, a message was loaded while the thread was joining, and hence not deleted sometimes and its reference blocked the rd_kafka_destroy().
Issue solved.
Thank you