admin. test topic creation.

Author: j7nw4r

tests/admin.rs

@@ -0,0 +1,48 @@
+use rdkafka::admin::{AdminOptions, NewTopic, TopicReplication};
+use crate::utils::containers::KafkaContext;
+use crate::utils::logging::init_test_logger;
+use crate::utils::rand::rand_test_topic;
+
+#[path = "utils/mod.rs"]
+mod utils;
+
+#[tokio::test]
+pub async fn test_topic_creation() {
+    init_test_logger();
+
+    // Get Kafka container context.
+    let kafka_context_result = KafkaContext::shared().await;
+    let Ok(kafka_context) = kafka_context_result else {
+        panic!(
+            "could not create kafka context: {}",
+            kafka_context_result.unwrap_err()
+        );
+    };
+    let test_topic_name = rand_test_topic("testing-topic");
+
+    let admin_client_result = utils::admin::create_admin(&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 {
+        panic!(
+            "could not create new topics: {}",
+            create_topics_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

@@ -3,9 +3,10 @@ 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 rdkafka::consumer::{BaseConsumer, Consumer};
-use rdkafka::error::KafkaError;
-use rdkafka::producer::{BaseRecord, Producer};
+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;
@@ -16,52 +17,46 @@ mod utils;
 pub async fn test_basic_produce() {
     init_test_logger();
 
-    let kafka_context_result = KafkaContext::new().await;
-    let Ok(_kafka_context) = kafka_context_result else {
+    // Get Kafka container context.
+    let kafka_context_result = KafkaContext::shared().await;
+    let Ok(kafka_context) = kafka_context_result else {
         panic!(
             "could not create kafka context: {}",
             kafka_context_result.unwrap_err()
         );
     };
+    let test_topic_name = rand_test_topic("testing-topic");
 
-    let bootstrap_servers_result = _kafka_context.bootstrap_servers().await;
-    let Ok(bootstrap_servers) = bootstrap_servers_result else {
+    let consumer_result = create_consumer(&kafka_context.bootstrap_servers, &test_topic_name).await;
+    let Ok(consumer) = consumer_result else {
         panic!(
-            "could not create bootstrap servers: {}",
-            bootstrap_servers_result.unwrap_err()
+            "could not create consumer: {}",
+            consumer_result.unwrap_err()
         );
     };
-    let test_topic = rand_test_topic("testing-topic");
-
-    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 {
+    let create_producer_result = create_producer(&kafka_context.bootstrap_servers).await;
+    let Ok(base_producer) = create_producer_result else {
         panic!(
-            "could not create based_producer: {}",
-            base_producer_result.unwrap_err()
+            "could not create base producer: {}",
+            create_producer_result.unwrap_err()
         );
     };
 
-    let record = BaseRecord::to(&test_topic) // destination topic
+    let record = BaseRecord::to(&test_topic_name) // destination topic
         .key(&[1, 2, 3, 4]) // message key
         .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 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!(
@@ -82,13 +77,13 @@ pub async fn test_basic_produce() {
 }
 
 async fn create_consumer(
-    bootstrap_servers: String,
-    test_topic: String,
+    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("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");
@@ -104,5 +99,18 @@ async fn create_consumer(
         )
     };
 
-    unimplemented!("unimplemented");
+    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,6 +16,7 @@ use rdkafka::{ClientConfig, Message, Timestamp};
 use crate::utils::rand::*;
 use crate::utils::*;
 
+mod admin;
 mod utils;
 
 fn create_base_consumer(

tests/utils/admin.rs

@@ -0,0 +1,13 @@
+use anyhow::Context;
+use rdkafka::admin::AdminClient;
+use rdkafka::client::DefaultClientContext;
+use rdkafka::config::FromClientConfig;
+use rdkafka::ClientConfig;
+
+pub async fn create_admin(
+    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")
+}

tests/utils/containers.rs

@@ -9,6 +9,7 @@ use tokio::sync::OnceCell;
 
 pub struct KafkaContext {
     kafka_node: ContainerAsync<Kafka>,
+    pub bootstrap_servers: String,
 }
 
 impl Debug for KafkaContext {
@@ -22,31 +23,31 @@ impl KafkaContext {
         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::<Arc<KafkaContext>, anyhow::Error>(Arc::new(KafkaContext { kafka_node }))
-            })
+            .get_or_try_init(init)
             .await
             .context("Failed to initialize Kafka shared instance")
             .map(Arc::clone)
     }
+}
 
-    pub async fn bootstrap_servers(&self) -> anyhow::Result<String> {
-        let kafka_host = self
-            .kafka_node
-            .get_host()
-            .await
-            .context("Failed to get Kafka host")?;
-        let kafka_port = self
-            .kafka_node
-            .get_host_port_ipv4(ContainerPort::Tcp(9092))
-            .await?;
-        Ok(format!("{}:{}", kafka_host.to_string(), kafka_port))
-    }
+async fn init() -> anyhow::Result<Arc<KafkaContext>> {
+    let kafka_node = Kafka::default()
+        .start()
+        .await
+        .context("Failed to start Kafka")?;
+
+    let kafka_host = kafka_node
+        .get_host()
+        .await
+        .context("Failed to get Kafka host")?;
+    let kafka_port = kafka_node
+        .get_host_port_ipv4(ContainerPort::Tcp(9092))
+        .await?;
+
+    Ok::<Arc<KafkaContext>, anyhow::Error>(Arc::new(KafkaContext {
+        kafka_node,
+        bootstrap_servers: format!("{}:{}", kafka_host, kafka_port),
+    }))
 }
 
 #[tokio::test]
@@ -59,13 +60,9 @@ pub async fn test_kafka_context_works() {
         );
     };
 
-    let bootstrap_servers_result = kafka_context.bootstrap_servers().await;
-    let Ok(bootstrap_servers) = bootstrap_servers_result else {
-        panic!(
-            "Failed to get bootstrap servers: {}",
-            bootstrap_servers_result.unwrap_err()
-        );
-    };
-
-    assert_ne!(bootstrap_servers.len(), 0, "Bootstrap servers empty");
+    assert_ne!(
+        kafka_context.bootstrap_servers.len(),
+        0,
+        "Bootstrap servers empty"
+    );
 }

tests/utils/mod.rs

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