From 6305727c994dc62b077b6dba8eee63b8c23f1518 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Marko=20Milenkovi=C4=87?=
 <milenkovicm@users.noreply.github.com>
Date: Tue, 30 Apr 2024 09:57:08 +0100
Subject: [PATCH] code and doc cleanup

---
 README.md             | 13 +++++-----
 tests/common/mod.rs   | 58 ++-----------------------------------------
 tests/create_topic.rs |  2 +-
 tests/redpanda.rs     |  2 +-
 4 files changed, 11 insertions(+), 64 deletions(-)

diff --git a/README.md b/README.md
index a2f09e2..8dd48d2 100644
--- a/README.md
+++ b/README.md
@@ -6,10 +6,10 @@
 
 Unofficial testcontainer for [Redpanda](https://redpanda.com).
 
-> [!NOTE]
->
-> - version 0.2.x supports `testcontainer` 0.16
-> - version 0.1.x supports `testcontainer` 0.15
+Note:
+
+- version `0.2.x` supports `testcontainer` `0.16`
+- version `0.1.x` supports `testcontainer` `0.15`
 
 Add dependency:
 
@@ -38,5 +38,6 @@ async fn main() {
 }
 ```
 
-> [!WARNING]  
-> It will use default kafka ports and only one test can  at any time on given host.
+Limitations:
+
+- It will use default kafka ports and only single test can run  on given host.
diff --git a/tests/common/mod.rs b/tests/common/mod.rs
index cbe34ad..42e3fbf 100644
--- a/tests/common/mod.rs
+++ b/tests/common/mod.rs
@@ -2,9 +2,8 @@
 //! # Common For tests
 //!
 
-#![allow(dead_code)]
+//#![allow(dead_code)]
 
-use rdkafka::admin::{AdminClient, AdminOptions, NewTopic, TopicReplication};
 use rdkafka::client::ClientContext;
 use rdkafka::config::ClientConfig;
 use rdkafka::consumer::ConsumerContext;
@@ -21,18 +20,6 @@ use std::time::Duration;
 /// Taken from https://github.com/fede1024/rust-rdkafka/blob/master/tests/utils.rs with some slight modifications and updates
 /// credit to rdkafka
 
-pub fn rand_test_topic() -> String {
-    format!("__test_{}", random_topic_name())
-}
-
-pub fn rand_test_group() -> String {
-    format!("__test_{}", random_topic_name())
-}
-
-pub fn rand_test_transactional_id() -> String {
-    format!("__test_{}", random_topic_name())
-}
-
 pub fn get_bootstrap_server() -> String {
     env::var("KAFKA_HOST").unwrap_or_else(|_| "localhost:9092".to_owned())
 }
@@ -45,21 +32,6 @@ impl ClientContext for ProducerTestContext {
     fn stats(&self, _: Statistics) {} // Don't print stats
 }
 
-pub async fn create_topic(name: &str, partitions: i32) {
-    let client: AdminClient<_> = consumer_config("create_topic", None)
-        .into_iter()
-        .collect::<ClientConfig>()
-        .create()
-        .unwrap();
-    client
-        .create_topics(
-            &[NewTopic::new(name, partitions, TopicReplication::Fixed(1))],
-            &AdminOptions::new(),
-        )
-        .await
-        .unwrap();
-}
-
 /// Produce the specified count of messages to the topic and partition specified. A map
 /// of (partition, offset) -> message id will be returned. It panics if any error is encountered
 /// while populating the topic.
@@ -84,7 +56,7 @@ where
         .set("bootstrap.servers", get_bootstrap_server().as_str())
         .set("statistics.interval.ms", "500")
         .set("api.version.request", "true")
-        .set("debug", "all")
+        //.set("debug", "all")
         .set("message.timeout.ms", "10000")
         .create_with_context::<ProducerTestContext, FutureProducer<_>>(prod_context)
         .expect("Producer creation error");
@@ -151,32 +123,6 @@ impl ConsumerContext for ConsumerTestContext {
     }
 }
 
-pub fn consumer_config<'a>(
-    group_id: &'a str,
-    config_overrides: Option<HashMap<&'a str, &'a str>>,
-) -> HashMap<String, String> {
-    let mut config: HashMap<String, String> = HashMap::new();
-
-    config.insert("group.id".into(), group_id.into());
-    config.insert("client.id".into(), "datafusion-streams".into());
-    config.insert("bootstrap.servers".into(), get_bootstrap_server());
-    config.insert("enable.partition.eof".into(), "true".into());
-    config.insert("session.timeout.ms".into(), "6000".into());
-    config.insert("enable.auto.commit".into(), "true".into());
-    config.insert("statistics.interval.ms".into(), "500".into());
-    config.insert("api.version.request".into(), "true".into());
-    config.insert("debug".into(), "all".into());
-    config.insert("auto.offset.reset".into(), "earliest".into());
-
-    if let Some(overrides) = config_overrides {
-        for (key, value) in overrides {
-            config.insert(key.into(), value.into());
-        }
-    }
-
-    config
-}
-
 #[cfg(test)]
 #[ctor::ctor]
 fn init() {
diff --git a/tests/create_topic.rs b/tests/create_topic.rs
index 40b97d7..5ae4fcf 100644
--- a/tests/create_topic.rs
+++ b/tests/create_topic.rs
@@ -15,7 +15,7 @@ mod test {
 
         // if topic has only one partition this part is optional
         // it will be automatically created when client connects
-        let test_topic_name = "test_topic";
+        let test_topic_name = &random_topic_name();
         server_node.exec(Redpanda::cmd_create_topic(test_topic_name, 3)).await;
 
         info!("bootstrap servers: {}", bootstrap_servers);
diff --git a/tests/redpanda.rs b/tests/redpanda.rs
index e4fd550..76194c0 100644
--- a/tests/redpanda.rs
+++ b/tests/redpanda.rs
@@ -21,7 +21,7 @@ mod test {
         std::env::set_var("KAFKA_HOST", &bootstrap_servers);
 
         assert!(bootstrap_servers.len() > 10);
-        let test_topic_name = "test_topic";
+        let test_topic_name = random_topic_name();
 
         log::info!("populating topic: [{}] ...", test_topic_name);
         populate_topic(&test_topic_name, 10, &value_fn, &key_fn, Some(0), None).await;