Add destroy call when leader epoch is stale (#4429)

and partition is in state active

Author: emasab

CHANGELOG.md

@@ -4,6 +4,8 @@ librdkafka v2.2.1 is a maintenance release:
 
  * Added Topic id to the metadata response which is part of the [KIP-516](https://cwiki.apache.org/confluence/display/KAFKA/KIP-516%3A+Topic+Identifiers)
  * Fixed ListConsumerGroupOffsets not fetching offsets for all the topics in a group with Apache Kafka version below 2.4.0.
+ * Add missing destroy that leads to leaking partition structure memory when there
+   are partition leader changes and a stale leader epoch is received (#4429).
 
 
 

src/rdkafka_topic.c

@@ -677,6 +677,7 @@ static int rd_kafka_toppar_leader_update(rd_kafka_topic_t *rkt,
                              rktp->rktp_leader_epoch);
                 if (rktp->rktp_fetch_state == RD_KAFKA_TOPPAR_FETCH_ACTIVE) {
                         rd_kafka_toppar_unlock(rktp);
+                        rd_kafka_toppar_destroy(rktp); /* from get() */
                         return 0;
                 }
         }