From 05652b33a5b31b9a8f067e63ee51d10010d979fe Mon Sep 17 00:00:00 2001 From: GeorgeBerdovskiy Date: Mon, 18 May 2026 10:17:46 -0700 Subject: [PATCH 1/2] Drop Native Config on `rd_kafka_new` Error --- src/client.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/client.rs b/src/client.rs index fd3fded5f..448366997 100644 --- a/src/client.rs +++ b/src/client.rs @@ -268,6 +268,7 @@ impl Client { trace!("Create new librdkafka client {:p}", client_ptr); if client_ptr.is_null() { + drop(native_config); return Err(KafkaError::ClientCreation(err_buf.to_string())); } From 37990cf7a285652f7ecd5842cf97cd857bd91bb4 Mon Sep 17 00:00:00 2001 From: GeorgeBerdovskiy Date: Mon, 18 May 2026 10:37:32 -0700 Subject: [PATCH 2/2] Fix Ownership Issue --- src/client.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/client.rs b/src/client.rs index 448366997..5eed92517 100644 --- a/src/client.rs +++ b/src/client.rs @@ -257,7 +257,6 @@ impl Client { native_config.set("log.queue", "true")?; let client_ptr = unsafe { - let native_config = ManuallyDrop::new(native_config); rdsys::rd_kafka_new( rd_kafka_type, native_config.ptr(), @@ -268,10 +267,12 @@ impl Client { trace!("Create new librdkafka client {:p}", client_ptr); if client_ptr.is_null() { - drop(native_config); return Err(KafkaError::ClientCreation(err_buf.to_string())); } + // Let Rust drop `native_config` on failure and only `forget` after `rd_kafka_new` succeeds + std::mem::forget(native_config); + let ret = unsafe { rdsys::rd_kafka_set_log_queue(client_ptr, rdsys::rd_kafka_queue_get_main(client_ptr)) };