test: converting test suite to TestContainers

Cargo.toml

@@ -42,6 +42,8 @@ rand = "0.9.1"
 regex = "1.11.1"
 smol = "2.0.2"
 tokio = { version = "1.18", features = ["macros", "rt-multi-thread", "time"] }
+testcontainers-modules = { version = "0.13.0", features = ["kafka"] }
+anyhow = { version = "1.0.100" }
 
 # These features are re-exports of the features that the rdkafka-sys crate
 # provides. See the rdkafka-sys documentation for details.

src/consumer/base_consumer.rs

@@ -1,6 +1,7 @@
 //! Low-level consumers.
 
 use std::ffi::{CStr, CString};
+use std::fmt;
 use std::mem::ManuallyDrop;
 use std::os::raw::c_void;
 use std::ptr;
@@ -41,6 +42,20 @@ where
     nonempty_callback: Option<Box<Box<dyn Fn() + Send + Sync>>>,
 }
 
+impl<C> fmt::Debug for BaseConsumer<C>
+where
+    C: ConsumerContext,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("BaseConsumer")
+            .field("native_ptr", &self.client.native_ptr())
+            .field("queue", &self.queue)
+            .field("group_id", &self.group_id)
+            .field("has_nonempty_callback", &self.nonempty_callback.is_some())
+            .finish()
+    }
+}
+
 impl FromClientConfig for BaseConsumer {
     fn from_config(config: &ClientConfig) -> KafkaResult<BaseConsumer> {
         BaseConsumer::from_config_and_context(config, DefaultConsumerContext)

src/producer/base_producer.rs

@@ -42,6 +42,7 @@
 //! should wait and try again.
 
 use std::ffi::{CStr, CString};
+use std::fmt;
 use std::marker::PhantomData;
 use std::mem;
 use std::os::raw::c_void;
@@ -340,6 +341,19 @@ where
     _partitioner: PhantomData<Part>,
 }
 
+impl<C, Part> fmt::Debug for BaseProducer<C, Part>
+where
+    Part: Partitioner,
+    C: ProducerContext<Part>,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("BaseProducer")
+            .field("native_ptr", &self.native_ptr())
+            .field("queue", &self.queue)
+            .finish()
+    }
+}
+
 impl<C, Part> BaseProducer<C, Part>
 where
     Part: Partitioner,