fix: memory leak in incremental assign (#35)

* fix: memory leak in incremental assign

* Update src/kafka-consumer.cc

Co-authored-by: Milind L <[email protected]>

* docs: add comment about delete toppars

---------

Co-authored-by: Milind L <[email protected]>

Author: martijnimhoff

src/kafka-consumer.cc

@@ -245,12 +245,15 @@ Baton KafkaConsumer::IncrementalUnassign(std::vector<RdKafka::TopicPartition*> p
 
   RdKafka::Error* error = consumer->incremental_unassign(partitions);
 
+  std::vector<RdKafka::TopicPartition*> delete_partitions;
+
   if (error == NULL) {
     // For now, use two for loops. Make more efficient if needed at a later point.
     for (unsigned int i = 0; i < partitions.size(); i++) {
       for (unsigned int j = 0; j < m_partitions.size(); j++) {
         if (partitions[i]->partition() == m_partitions[j]->partition() &&
             partitions[i]->topic() == m_partitions[j]->topic()) {
+          delete_partitions.push_back(m_partitions[j]);
           m_partitions.erase(m_partitions.begin() + j);
           m_partition_cnt--;
           break;
@@ -260,7 +263,11 @@ Baton KafkaConsumer::IncrementalUnassign(std::vector<RdKafka::TopicPartition*> p
   }
 
   // Destroy the old list of partitions since we are no longer using it
-  RdKafka::TopicPartition::destroy(m_partitions);
+  RdKafka::TopicPartition::destroy(delete_partitions);
+
+  // Destroy the partition args since those are only used to lookup the partitions
+  // that needed to be deleted.
+  RdKafka::TopicPartition::destroy(partitions);
 
   return rdkafkaErrorToBaton(error);
 }