(WITHOUT object leak) RdKafka::Consumer destructor hangs forever (when kafka cluster re-starts taking place)

[aKumara123]

Description

THERE IS NO OBJECT LEAK - program to re-create the issue attached here.
RdKafka::Consumer destructor thread hangs due to broker thread rkb->rkb_refcnt being NOT zero.
This happens during kafka cluster re-starts.

Operations like reading the last msg, query_watermark_offsets done, while Kafka cluster is being killed & re-started.

How to reproduce

Sample program attached.

How to Compile

In a Linux environment :

1. Extract the attached sample_prog_to_re_create_issue.tar.gz
2. Go to it's 'build' directory
3. Set environment variables KAFKA_INC_PATH(point to librdkafka header files containing directory - both C and C++) and KAFKA_LINK_PATH(point to librdkafka shared libs containing directory)
4. type 'cmake ..'
5. type 'make'

This will build the binary LibrdKafkaIssueTest.

How to Run

Create 2 Kafka topics with topic partition count=4 for each.(3 broker cluster with replication factor=2 used for my re-creation)
Keep one topic empty and write a single message to each of the partition in other topic
Use bash command to run Kafka cluster kill-restart loop : 'while true; do [stop_the_kafka_cluster]; sleep 10; [re-start-kafka-cluster-WITHOUT-cleaning-data]; sleep 180; done'
Invoke LibrdKafkaIssueTest as :
'LibrdKafkaIssueTest 175 [kafka boost trap servers] [empty topic] [topic with 1 msg per partition] 0'

Run tool for about 10 minutes.
When issue is re-created it will show msg (in Color RED), 'Thread Index=[index] is stuck in Consumer Destroy for [stuck time]s'.

Checklist

• librdkafka version (release number or git tag): 1.9.2 (issue present in 1.6 too)

• Apache Kafka version: 2.7.0

• librdkafka client configuration:All default values used

• Operating system: Red Hat 8.4 x64 bit (Issue re-created on Red Hat Linux 7.9 too)

• Provide logs : attached
  -- gdb.txt
  -- librd kafka log output librdkafka_output.zip - Kindly Please note that IP and Port names are substituted.
  -- stack stack_3323083.zip
  -- stderr output : NOTE librdkafka was built with ENABLE_REFCNT_DEBUG stderr_LibrdKafkaIssueTest_1.zip

• [] Provide broker log excerpts : Log sizes are large, hence NOT attached. Please let me know if this is essential, I will re-create the issue and attach all logs

• Critical issue : No

[rickif]

This is similar to #3954.