basic producer test works.

Author: j7nw4r

tests/admin.rs

@@ -1,4 +1,4 @@
-use rdkafka::admin::{AdminOptions, NewTopic, TopicReplication};
+use crate::utils::admin::create_topic;
 use crate::utils::containers::KafkaContext;
 use crate::utils::logging::init_test_logger;
 use crate::utils::rand::rand_test_topic;
@@ -20,29 +20,20 @@ pub async fn test_topic_creation() {
     };
     let test_topic_name = rand_test_topic("testing-topic");
 
-    let admin_client_result = utils::admin::create_admin(&kafka_context.bootstrap_servers).await;
+    let admin_client_result =
+        utils::admin::create_admin_client(&kafka_context.bootstrap_servers).await;
     let Ok(admin_client) = admin_client_result else {
         panic!(
             "could not create admin client: {}",
             admin_client_result.unwrap_err()
         );
     };
 
-    let new_topic = NewTopic::new(&test_topic_name, 1, TopicReplication::Fixed(1));
-    let admin_opts = AdminOptions::new();
-    let create_topics_result = admin_client
-        .create_topics(vec![&new_topic], &admin_opts)
-        .await;
-    let Ok(create_topics) = create_topics_result else {
+    let create_topic_result = create_topic(&admin_client, &test_topic_name).await;
+    if create_topic_result.is_err() {
         panic!(
-            "could not create new topics: {}",
-            create_topics_result.unwrap_err()
+            "could not create topic: {}",
+            create_topic_result.unwrap_err()
         );
     };
-
-    for topic_result in create_topics {
-        if topic_result.is_err() {
-            panic!("failed to create topic: {:?}", topic_result.unwrap_err());
-        };
-    }
 }

tests/producer.rs

@@ -1,23 +1,19 @@
-use crate::utils::consumer;
+use crate::utils::admin::{create_admin_client, create_topic};
+use crate::utils::consumer::{create_consumer, poll_x_times_for_messages};
 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_group, rand_test_topic};
-use anyhow::Context;
-use rdkafka::config::FromClientConfig;
-use rdkafka::consumer::BaseConsumer;
-use rdkafka::producer::{BaseProducer, BaseRecord, Producer};
-use rdkafka::util::Timeout;
-use rdkafka::{ClientConfig, Message};
-use std::time::Duration;
+use crate::utils::producer::{create_producer, poll_and_flush};
+use crate::utils::rand::rand_test_topic;
+use rdkafka::producer::BaseRecord;
+use rdkafka::Message;
 
+#[path = "utils/mod.rs"]
 mod utils;
 
 #[tokio::test]
 pub async fn test_basic_produce() {
     init_test_logger();
 
-    // Get Kafka container context.
     let kafka_context_result = KafkaContext::shared().await;
     let Ok(kafka_context) = kafka_context_result else {
         panic!(
@@ -27,13 +23,29 @@ pub async fn test_basic_produce() {
     };
     let test_topic_name = rand_test_topic("testing-topic");
 
+    let admin_client_result = create_admin_client(&kafka_context.bootstrap_servers).await;
+    let Ok(admin_client) = admin_client_result else {
+        panic!(
+            "could not create admin client: {}",
+            admin_client_result.unwrap_err()
+        );
+    };
+    let create_topic_result = create_topic(&admin_client, &test_topic_name).await;
+    if create_topic_result.is_err() {
+        panic!(
+            "could not create topic: {}",
+            create_topic_result.unwrap_err()
+        );
+    }
+
     let consumer_result = create_consumer(&kafka_context.bootstrap_servers, &test_topic_name).await;
     let Ok(consumer) = consumer_result else {
         panic!(
             "could not create consumer: {}",
             consumer_result.unwrap_err()
         );
     };
+
     let create_producer_result = create_producer(&kafka_context.bootstrap_servers).await;
     let Ok(base_producer) = create_producer_result else {
         panic!(
@@ -50,20 +62,19 @@ pub async fn test_basic_produce() {
     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);
+    if poll_and_flush(&base_producer).is_err() {
+        panic!("could not poll and flush base producer")
     }
 
-    let Some(next_message_result) = consumer.poll(Duration::from_secs(2)) else {
-        panic!("there is no next message on the topic: {}", test_topic_name);
-    };
-    let Ok(borrowed_next_message) = next_message_result else {
-        panic!(
-            "could not get next message from based_consumer: {}",
-            next_message_result.unwrap_err()
-        );
+    let messages_result = poll_x_times_for_messages(&consumer, 10).await;
+    let Ok(messages) = messages_result else {
+        panic!("could not get messages from consumer");
     };
+    if messages.len() != 1 {
+        panic!("expected exactly one message");
+    }
+    let borrowed_next_message = messages.get(0).unwrap();
+
     let owned_next_message = borrowed_next_message.detach();
     let Some(message_payload) = owned_next_message.payload() else {
         panic!("message payload is empty");
@@ -75,42 +86,3 @@ pub async fn test_basic_produce() {
 
     assert!(message_string.contains("content"));
 }
-
-async fn create_consumer(
-    bootstrap_servers: &str,
-    test_topic: &str,
-) -> 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()
-        )
-    };
-
-    Ok(base_consumer)
-}
-
-async fn create_producer(bootstrap_servers: &str) -> anyhow::Result<BaseProducer> {
-    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()
-        );
-    };
-    Ok(base_producer)
-}

tests/test_low_consumers.rs

@@ -16,7 +16,6 @@ use rdkafka::{ClientConfig, Message, Timestamp};
 use crate::utils::rand::*;
 use crate::utils::*;
 
-mod admin;
 mod utils;
 
 fn create_base_consumer(

tests/utils/admin.rs

@@ -1,13 +1,34 @@
-use anyhow::Context;
-use rdkafka::admin::AdminClient;
+use anyhow::{bail, Context};
+use rdkafka::admin::{AdminClient, AdminOptions, NewTopic, TopicReplication};
 use rdkafka::client::DefaultClientContext;
 use rdkafka::config::FromClientConfig;
 use rdkafka::ClientConfig;
 
-pub async fn create_admin(
+pub async fn create_admin_client(
     bootstrap_servers: &str,
 ) -> anyhow::Result<AdminClient<DefaultClientContext>> {
     let mut admin_client_config = ClientConfig::default();
     admin_client_config.set("bootstrap.servers", bootstrap_servers);
     AdminClient::from_config(&admin_client_config).context("error creating admin client")
 }
+
+pub async fn create_topic(
+    admin_client: &AdminClient<DefaultClientContext>,
+    topic_name: &'_ str,
+) -> anyhow::Result<String> {
+    let topic_results = admin_client
+        .create_topics(&new_topic_vec(&topic_name), &AdminOptions::default())
+        .await
+        .context("error creating topics")?;
+    for topic_result in topic_results {
+        if topic_result.is_err() {
+            bail!("failed to create topic: {:?}", topic_result.unwrap_err());
+        };
+    }
+    Ok(topic_name.to_string())
+}
+
+pub fn new_topic_vec(topic_name: &'_ str) -> Vec<NewTopic<'_>> {
+    let new_topic = NewTopic::new(&topic_name, 1, TopicReplication::Fixed(1));
+    vec![new_topic]
+}

tests/utils/consumer.rs

@@ -1,7 +1,37 @@
+use crate::utils::consumer;
+use crate::utils::rand::rand_test_group;
 use anyhow::Context;
 use rdkafka::config::FromClientConfig;
 use rdkafka::consumer::{BaseConsumer, Consumer};
+use rdkafka::message::BorrowedMessage;
 use rdkafka::ClientConfig;
+use std::time::Duration;
+
+pub async fn create_consumer(
+    bootstrap_servers: &str,
+    test_topic: &str,
+) -> 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()
+        )
+    };
+
+    Ok(base_consumer)
+}
 
 pub async fn create_subscribed_base(
     client_config: ClientConfig,
@@ -15,3 +45,27 @@ pub async fn create_subscribed_base(
 
     Ok(base_consumer)
 }
+
+pub async fn poll_x_times_for_messages(
+    consumer: &BaseConsumer,
+    times_to_poll: i32,
+) -> anyhow::Result<Vec<BorrowedMessage<'_>>> {
+    let mut borrowed_messages: Vec<BorrowedMessage> = Vec::new();
+
+    for _ in 0..times_to_poll {
+        let Some(next_message_result) = consumer.poll(Duration::from_secs(2)) else {
+            continue;
+        };
+
+        let Ok(borrowed_next_message) = next_message_result else {
+            panic!(
+                "could not get next message from based_consumer: {}",
+                next_message_result.unwrap_err()
+            );
+        };
+        borrowed_messages.push(borrowed_next_message);
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+
+    Ok(borrowed_messages)
+}

tests/utils/containers.rs

@@ -28,6 +28,24 @@ impl KafkaContext {
             .context("Failed to initialize Kafka shared instance")
             .map(Arc::clone)
     }
+
+    pub async fn std_out(&self) -> anyhow::Result<String> {
+        let std_out_byte_vec = self
+            .kafka_node
+            .stdout_to_vec()
+            .await
+            .context("Failed to get stdout")?;
+        Ok(String::from_utf8(std_out_byte_vec)?)
+    }
+
+    pub async fn std_err(&self) -> anyhow::Result<String> {
+        let std_err_byte_vec = self
+            .kafka_node
+            .stderr_to_vec()
+            .await
+            .context("Failed to get stderr")?;
+        Ok(String::from_utf8(std_err_byte_vec)?)
+    }
 }
 
 async fn init() -> anyhow::Result<Arc<KafkaContext>> {

tests/utils/producer.rs

@@ -1,6 +1,22 @@
+use anyhow::Context;
 use rdkafka::config::FromClientConfig;
-use rdkafka::producer::BaseProducer;
+use rdkafka::producer::{BaseProducer, Producer};
+use rdkafka::util::Timeout;
 use rdkafka::ClientConfig;
+use std::time::Duration;
+
+pub async fn create_producer(bootstrap_servers: &str) -> anyhow::Result<BaseProducer> {
+    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()
+        );
+    };
+    Ok(base_producer)
+}
 
 pub fn create_base_producer(config: &ClientConfig) -> anyhow::Result<BaseProducer> {
     let base_producer_result = BaseProducer::from_config(config);
@@ -10,6 +26,14 @@ pub fn create_base_producer(config: &ClientConfig) -> anyhow::Result<BaseProduce
             base_producer_result.unwrap_err()
         )
     };
-
     Ok(base_producer)
 }
+
+pub fn poll_and_flush(base_producer: &BaseProducer) -> anyhow::Result<()> {
+    for _ in 0..5 {
+        base_producer.poll(Duration::from_millis(100));
+    }
+    base_producer
+        .flush(Timeout::After(Duration::from_secs(10)))
+        .context("flush failed")
+}

tests/utils/rand.rs

@@ -2,7 +2,7 @@ use rand::distr::{Alphanumeric, SampleString};
 
 pub fn rand_test_topic(test_name: &str) -> String {
     let id = Alphanumeric.sample_string(&mut rand::rng(), 10);
-    format!("__{}_{}", test_name, id)
+    format!("{}_{}", test_name, id)
 }
 
 pub fn rand_test_group() -> String {