fix: rd_kafka_event_t is not destroyed

Author: BewareMyPower

create_topic.h

@@ -15,6 +15,7 @@
  */
 #pragma once
 
+#include "rk_event_wrapper.h"
 #include <array>
 #include <iostream>
 #include <librdkafka/rdkafka.h>
@@ -38,12 +39,12 @@ inline void create_topic(rd_kafka_t *rk, rd_kafka_queue_t *rkqu,
 
   rd_kafka_CreateTopics(rk, &rk_topic, 1, nullptr, rkqu);
 
-  if (auto event = rd_kafka_queue_poll(rkqu, -1 /* infinite timeout */);
-      rd_kafka_event_error(event)) {
-    std::cerr << "CreateTopics failed for " << topic << ": "
-              << rd_kafka_event_error_string(event);
-  } else {
+  try {
+    RdKafkaEvent::poll(rk, rkqu);
     std::cout << R"(Created topic ")" << topic << R"(" with )" << num_partitions
               << " partition" << (num_partitions == 1 ? "" : "s") << std::endl;
+  } catch (const std::runtime_error &e) {
+    std::cerr << "CreateTopics failed for " << topic << ": " << e.what();
+    return;
   }
 }

delete_topic.h

@@ -15,6 +15,7 @@
  */
 #pragma once
 
+#include "rk_event_wrapper.h"
 #include <iostream>
 #include <librdkafka/rdkafka.h>
 #include <memory>
@@ -31,12 +32,9 @@ inline void delete_topic(rd_kafka_t *rk, rd_kafka_queue_t *rkqu,
 
   rd_kafka_DeleteTopics(rk, &rk_topic, 1, nullptr, rkqu);
 
-  if (auto event = rd_kafka_queue_poll(rkqu, -1 /* infinite timeout */);
-      rd_kafka_event_error(event)) {
-    std::cerr << "DeleteTopics failed for " << topic << ": "
-              << rd_kafka_event_error_string(event);
-  } else {
-    auto result = rd_kafka_event_DeleteTopics_result(event);
+  try {
+    auto event = RdKafkaEvent::poll(rk, rkqu);
+    auto result = rd_kafka_event_DeleteTopics_result(event.handle());
     size_t cntp;
     auto topics = rd_kafka_DeleteTopics_result_topics(result, &cntp);
     if (cntp != 1) {
@@ -50,5 +48,8 @@ inline void delete_topic(rd_kafka_t *rk, rd_kafka_queue_t *rkqu,
       std::cerr << R"(Failed to delete topic ")" << topic << R"(": )" << error
                 << std::endl;
     }
+  } catch (const std::runtime_error &e) {
+    std::cerr << "DeleteTopics failed for " << topic << ": " << e.what()
+              << std::endl;
   }
 }

describe_topic.h

@@ -15,9 +15,10 @@
  */
 #pragma once
 
-#include <argparse/argparse.hpp>
+#include "rk_event_wrapper.h"
 #include <iostream>
 #include <librdkafka/rdkafka.h>
+#include <stdexcept>
 #include <string>
 
 inline void describe_topic(rd_kafka_t *rk, rd_kafka_queue_t *rkqu,
@@ -30,12 +31,9 @@ inline void describe_topic(rd_kafka_t *rk, rd_kafka_queue_t *rkqu,
 
   rd_kafka_DescribeTopics(rk, topic_names, nullptr, rkqu);
 
-  if (auto event = rd_kafka_queue_poll(rkqu, -1 /* infinite timeout */);
-      rd_kafka_event_error(event)) {
-    std::cerr << "DescribeTopics failed for " << topic << ": "
-              << rd_kafka_event_error_string(event);
-  } else {
-    auto result = rd_kafka_event_DescribeTopics_result(event);
+  try {
+    auto event = RdKafkaEvent::poll(rk, rkqu);
+    auto result = rd_kafka_event_DescribeTopics_result(event.handle());
     size_t result_topics_cnt;
     auto result_topics =
         rd_kafka_DescribeTopics_result_topics(result, &result_topics_cnt);
@@ -66,5 +64,8 @@ inline void describe_topic(rd_kafka_t *rk, rd_kafka_queue_t *rkqu,
         }
       }
     }
+  } catch (const std::runtime_error &e) {
+    std::cerr << "DescribeTopics failed for " << topic << ": " << e.what()
+              << std::endl;
   }
 }

rk_event_wrapper.h

@@ -0,0 +1,49 @@
+/**
+ * Copyright 2025 Yunze Xu
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <librdkafka/rdkafka.h>
+#include <stdexcept>
+
+class RdKafkaEvent final {
+public:
+  static RdKafkaEvent poll(rd_kafka_t *rk, rd_kafka_queue_t *rkqu) {
+    auto event = rd_kafka_queue_poll(rkqu, -1 /* infinite timeout */);
+    if (rd_kafka_event_error(event)) {
+      throw std::runtime_error(rd_kafka_event_error_string(event));
+    }
+    return RdKafkaEvent(event);
+  }
+
+  RdKafkaEvent(const RdKafkaEvent &) = delete;
+  RdKafkaEvent(RdKafkaEvent &&rhs) noexcept : event_(rhs.event_) {
+    rhs.event_ = nullptr;
+  }
+
+  ~RdKafkaEvent() {
+    if (event_ != nullptr) {
+      rd_kafka_event_destroy(event_);
+    }
+  }
+
+  // Get the underlying C handle for rdkafka's C APIs to use
+  auto handle() const noexcept { return event_; }
+
+private:
+  rd_kafka_event_t *event_;
+
+  RdKafkaEvent(rd_kafka_event_t *event) : event_(event) {}
+};