Various updates. KafkaContext is now shared.

Author: j7nw4r

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)

tests/producer.rs

@@ -1,9 +1,14 @@
-use rdkafka::ClientConfig;
+use crate::utils::consumer;
 use crate::utils::containers::KafkaContext;
 use crate::utils::logging::init_test_logger;
 use crate::utils::producer::create_base_producer;
-use crate::utils::rand::rand_test_topic;
-use rdkafka::producer::BaseRecord;
+use crate::utils::rand::{rand_test_group, rand_test_topic};
+use rdkafka::consumer::{BaseConsumer, Consumer};
+use rdkafka::error::KafkaError;
+use rdkafka::producer::{BaseRecord, Producer};
+use rdkafka::util::Timeout;
+use rdkafka::{ClientConfig, Message};
+use std::time::Duration;
 
 mod utils;
 
@@ -21,27 +26,83 @@ pub async fn test_basic_produce() {
 
     let bootstrap_servers_result = _kafka_context.bootstrap_servers().await;
     let Ok(bootstrap_servers) = bootstrap_servers_result else {
-        panic!("could not create bootstrap servers: {}", bootstrap_servers_result.unwrap_err());
+        panic!(
+            "could not create bootstrap servers: {}",
+            bootstrap_servers_result.unwrap_err()
+        );
     };
-    let mut client_config = ClientConfig::default();
-    client_config.set("bootstrap.servers", bootstrap_servers);
+    let test_topic = rand_test_topic("testing-topic");
 
-    let base_producer_result = create_base_producer(&client_config);
+    create_consumer(bootstrap_servers, test_topic);
+
+    let mut producer_client_config = ClientConfig::default();
+    producer_client_config.set("bootstrap.servers", &bootstrap_servers);
+
+    let base_producer_result = create_base_producer(&producer_client_config);
     let Ok(base_producer) = base_producer_result else {
         panic!(
             "could not create based_producer: {}",
             base_producer_result.unwrap_err()
         );
     };
 
-    let test_topic = rand_test_topic("testing-topic");
     let record = BaseRecord::to(&test_topic) // destination topic
         .key(&[1, 2, 3, 4]) // message key
-        .payload("content") // message payload
-        .partition(5);
+        .payload("content"); // message payload
 
     let send_result = base_producer.send(record);
     if send_result.is_err() {
         panic!("could not produce record: {:?}", send_result.unwrap_err());
     }
+
+    let flush_result = base_producer.flush(Timeout::After(Duration::from_secs(10)));
+    if let Err(flush_error) = flush_result {
+        panic!("timed out waiting for producer flush: {}", flush_error);
+    }
+
+    let Some(next_message_result) = base_consumer.poll(Duration::from_secs(2)) else {
+        panic!("there is no next message on the topic: {}", test_topic);
+    };
+    let Ok(borrowed_next_message) = next_message_result else {
+        panic!(
+            "could not get next message from based_consumer: {}",
+            next_message_result.unwrap_err()
+        );
+    };
+    let owned_next_message = borrowed_next_message.detach();
+    let Some(message_payload) = owned_next_message.payload() else {
+        panic!("message payload is empty");
+    };
+    let message_string_result = String::from_utf8(message_payload.to_vec());
+    let Ok(message_string) = message_string_result else {
+        panic!("message payload is not valid UTF-8");
+    };
+
+    assert!(message_string.contains("content"));
+}
+
+async fn create_consumer(
+    bootstrap_servers: String,
+    test_topic: String,
+) -> anyhow::Result<BaseConsumer> {
+    let mut consumer_client_config = ClientConfig::default();
+    consumer_client_config.set("group.id", rand_test_group());
+    consumer_client_config.set("client.id", "rdkafka_integration_test_client");
+    consumer_client_config.set("bootstrap.servers", &bootstrap_servers);
+    consumer_client_config.set("enable.partition.eof", "false");
+    consumer_client_config.set("session.timeout.ms", "6000");
+    consumer_client_config.set("enable.auto.commit", "false");
+    consumer_client_config.set("debug", "all");
+    consumer_client_config.set("auto.offset.reset", "earliest");
+
+    let base_consumer_result =
+        consumer::create_subscribed_base(consumer_client_config, &[&test_topic]).await;
+    let Ok(base_consumer) = base_consumer_result else {
+        panic!(
+            "could not create base consumer: {}",
+            base_consumer_result.unwrap_err()
+        )
+    };
+
+    unimplemented!("unimplemented");
 }

tests/utils/consumer.rs

@@ -0,0 +1,17 @@
+use anyhow::Context;
+use rdkafka::config::FromClientConfig;
+use rdkafka::consumer::{BaseConsumer, Consumer};
+use rdkafka::ClientConfig;
+
+pub async fn create_subscribed_base(
+    client_config: ClientConfig,
+    topics: &[&str],
+) -> anyhow::Result<BaseConsumer> {
+    let base_consumer =
+        BaseConsumer::from_config(&client_config).context("Failed to create consumer")?;
+    base_consumer
+        .subscribe(topics)
+        .context("Failed to subscribe to topic")?;
+
+    Ok(base_consumer)
+}

tests/utils/containers.rs

@@ -1,9 +1,11 @@
 use anyhow::Context;
 use std::fmt::Debug;
+use std::sync::Arc;
 use testcontainers_modules::kafka::apache::Kafka;
 use testcontainers_modules::testcontainers::core::ContainerPort;
 use testcontainers_modules::testcontainers::runners::AsyncRunner;
 use testcontainers_modules::testcontainers::ContainerAsync;
+use tokio::sync::OnceCell;
 
 pub struct KafkaContext {
     kafka_node: ContainerAsync<Kafka>,
@@ -16,13 +18,21 @@ impl Debug for KafkaContext {
 }
 
 impl KafkaContext {
-    pub async fn new() -> anyhow::Result<Self> {
-        let kafka_node = Kafka::default()
-            .start()
-            .await
-            .context("Failed to start Kafka")?;
+    pub async fn shared() -> anyhow::Result<Arc<Self>> {
+        static INSTANCE: OnceCell<Arc<KafkaContext>> = OnceCell::const_new();
+
+        INSTANCE
+            .get_or_try_init(|| async {
+                let kafka_node = Kafka::default()
+                    .start()
+                    .await
+                    .context("Failed to start Kafka")?;
 
-        Ok(Self { kafka_node })
+                Ok::<Arc<KafkaContext>, anyhow::Error>(Arc::new(KafkaContext { kafka_node }))
+            })
+            .await
+            .context("Failed to initialize Kafka shared instance")
+            .map(Arc::clone)
     }
 
     pub async fn bootstrap_servers(&self) -> anyhow::Result<String> {
@@ -41,7 +51,7 @@ impl KafkaContext {
 
 #[tokio::test]
 pub async fn test_kafka_context_works() {
-    let kafka_context_result = KafkaContext::new().await;
+    let kafka_context_result = KafkaContext::shared().await;
     let Ok(kafka_context) = kafka_context_result else {
         panic!(
             "Failed to get Kafka context: {}",

tests/utils/mod.rs

@@ -1,5 +1,6 @@
 #![allow(dead_code)]
 
+pub mod consumer;
 pub mod containers;
 pub mod logging;
 pub mod producer;

tests/utils/producer.rs

@@ -1,11 +1,15 @@
-use rdkafka::ClientConfig;
 use rdkafka::config::FromClientConfig;
 use rdkafka::producer::BaseProducer;
+use rdkafka::ClientConfig;
 
 pub fn create_base_producer(config: &ClientConfig) -> anyhow::Result<BaseProducer> {
     let base_producer_result = BaseProducer::from_config(config);
     let Ok(base_producer) = base_producer_result else {
-        anyhow::bail!("error creating base producer: {}", base_producer_result.unwrap_err())
+        anyhow::bail!(
+            "error creating base producer: {}",
+            base_producer_result.unwrap_err()
+        )
     };
+
     Ok(base_producer)
 }