fix: test_configs

Author: j7nw4r

tests/admin.rs

@@ -5,8 +5,8 @@ use crate::utils::rand::rand_test_topic;
 use crate::utils::{get_broker_version, KafkaVersion};
 use backon::{BlockingRetryable, ExponentialBuilder};
 use rdkafka::admin::{
-    AdminOptions, ConfigEntry, ConfigSource, NewPartitions, NewTopic, ResourceSpecifier,
-    TopicReplication,
+    AdminOptions, AlterConfig, ConfigEntry, ConfigSource, NewPartitions, NewTopic,
+    OwnedResourceSpecifier, ResourceSpecifier, TopicReplication,
 };
 use rdkafka::error::KafkaError;
 use rdkafka::producer::{FutureRecord, Producer};
@@ -541,3 +541,85 @@ async fn test_delete_records() {
     assert_eq!(lo, 5);
     assert_eq!(hi, 5);
 }
+
+#[tokio::test]
+async fn test_configs() {
+    init_test_logger();
+
+    // Get Kafka container context.
+    let kafka_context = KafkaContext::shared()
+        .await
+        .expect("could not create kafka context");
+
+    // Create admin client
+    let admin_client = utils::admin::create_admin_client(&kafka_context.bootstrap_servers)
+        .await
+        .expect("could not create admin client");
+    let opts = AdminOptions::new();
+    let broker = ResourceSpecifier::Broker(utils::BROKER_ID);
+
+    let res = admin_client
+        .describe_configs(&[broker], &opts)
+        .await
+        .expect("describe configs failed");
+    let config = &res[0].as_ref().expect("describe configs failed");
+    let orig_val = config
+        .get("log.flush.interval.messages")
+        .expect("original config entry missing")
+        .value
+        .as_ref()
+        .expect("original value missing");
+
+    let config = AlterConfig::new(broker).set("log.flush.interval.messages", "1234");
+    let res = admin_client
+        .alter_configs(&[config], &opts)
+        .await
+        .expect("alter configs failed");
+    assert_eq!(res, &[Ok(OwnedResourceSpecifier::Broker(utils::BROKER_ID))]);
+
+    let mut tries = 0;
+    loop {
+        let res = admin_client
+            .describe_configs(&[broker], &opts)
+            .await
+            .expect("describe configs failed");
+        let config = &res[0].as_ref().expect("describe configs failed");
+        let entry = config.get("log.flush.interval.messages");
+        let expected_entry = if get_broker_version(&kafka_context) < KafkaVersion(1, 1, 0, 0) {
+            // Pre-1.1, the AlterConfig operation will silently fail, and the
+            // config will remain unchanged, which I guess is worth testing.
+            ConfigEntry {
+                name: "log.flush.interval.messages".into(),
+                value: Some(orig_val.clone()),
+                source: ConfigSource::Default,
+                is_read_only: true,
+                is_default: true,
+                is_sensitive: false,
+            }
+        } else {
+            ConfigEntry {
+                name: "log.flush.interval.messages".into(),
+                value: Some("1234".into()),
+                source: ConfigSource::DynamicBroker,
+                is_read_only: false,
+                is_default: false,
+                is_sensitive: false,
+            }
+        };
+        if entry == Some(&expected_entry) {
+            break;
+        } else if tries >= 5 {
+            panic!("{:?} != {:?}", entry, Some(&expected_entry));
+        } else {
+            tries += 1;
+            tokio::time::sleep(Duration::from_secs(1)).await;
+        }
+    }
+
+    let config = AlterConfig::new(broker).set("log.flush.interval.ms", orig_val);
+    let res = admin_client
+        .alter_configs(&[config], &opts)
+        .await
+        .expect("alter configs failed");
+    assert_eq!(res, &[Ok(OwnedResourceSpecifier::Broker(utils::BROKER_ID))]);
+}

tests/producer.rs

@@ -2,7 +2,7 @@ use crate::utils::admin::{create_admin_client, create_topic};
 use crate::utils::consumer::{create_subscribed_base_consumer, poll_x_times_for_messages};
 use crate::utils::containers::KafkaContext;
 use crate::utils::logging::init_test_logger;
-use crate::utils::producer::{create_producer, send_record};
+use crate::utils::producer::base_producer::create_producer;
 use crate::utils::rand::rand_test_topic;
 use rdkafka::producer::BaseRecord;
 use rdkafka::Message;
@@ -58,7 +58,8 @@ pub async fn test_basic_produce() {
     let record = BaseRecord::to(&test_topic_name) // destination topic
         .key(&[1, 2, 3, 4]) // message key
         .payload("content"); // message payload
-    let send_record_result = send_record(&base_producer, record).await;
+    let send_record_result =
+        crate::utils::producer::base_producer::send_record(&base_producer, record).await;
     if send_record_result.is_err() {
         panic!("could not send record: {}", send_record_result.unwrap_err());
     }

tests/test_admin.rs

@@ -119,186 +119,6 @@ fn verify_delete(topic: &str) {
     .unwrap()
 }
 
-/// Test the admin client's delete records functionality.
-#[tokio::test]
-async fn test_delete_records() {
-    let producer = create_config().create::<FutureProducer<_>>().unwrap();
-    let admin_client = create_admin_client();
-    let timeout = Some(Duration::from_secs(1));
-    let opts = AdminOptions::new().operation_timeout(timeout);
-    let topic = rand_test_topic("test_delete_records");
-    let make_record = || FutureRecord::<str, str>::to(&topic).payload("data");
-
-    // Create a topic with a single partition.
-    admin_client
-        .create_topics(
-            &[NewTopic::new(&topic, 1, TopicReplication::Fixed(1))],
-            &opts,
-        )
-        .await
-        .expect("topic creation failed");
-
-    // Ensure that the topic begins with low and high water marks of 0.
-    let (lo, hi) = (|| producer.client().fetch_watermarks(&topic, 0, timeout))
-        .retry(ExponentialBuilder::default().with_max_delay(Duration::from_secs(5)))
-        .call()
-        .unwrap();
-    assert_eq!(lo, 0);
-    assert_eq!(hi, 0);
-
-    // Produce five messages to the topic.
-    for _ in 0..5 {
-        producer.send(make_record(), timeout).await.unwrap();
-    }
-
-    // Ensure that the high water mark has advanced to 5.
-    let (lo, hi) = producer
-        .client()
-        .fetch_watermarks(&topic, 0, timeout)
-        .unwrap();
-    assert_eq!(lo, 0);
-    assert_eq!(hi, 5);
-
-    // Delete the record at offset 0.
-    let mut tpl = TopicPartitionList::new();
-    tpl.add_partition_offset(&topic, 0, Offset::Offset(1))
-        .unwrap();
-    let res_tpl = admin_client.delete_records(&tpl, &opts).await.unwrap();
-    assert_eq!(res_tpl.count(), 1);
-    assert_eq!(res_tpl.elements()[0].topic(), topic);
-    assert_eq!(res_tpl.elements()[0].partition(), 0);
-    assert_eq!(res_tpl.elements()[0].offset(), Offset::Offset(1));
-    assert_eq!(res_tpl.elements()[0].error(), Ok(()));
-
-    // Ensure that the low water mark has advanced to 1.
-    let (lo, hi) = producer
-        .client()
-        .fetch_watermarks(&topic, 0, timeout)
-        .unwrap();
-    assert_eq!(lo, 1);
-    assert_eq!(hi, 5);
-
-    // Delete the record at offset 1 and also include an invalid partition in
-    // the request. The invalid partition should not cause the request to fail,
-    // but we should be able to see the per-partition error in the returned
-    // topic partition list.
-    let mut tpl = TopicPartitionList::new();
-    tpl.add_partition_offset(&topic, 0, Offset::Offset(2))
-        .unwrap();
-    tpl.add_partition_offset(&topic, 1, Offset::Offset(1))
-        .unwrap();
-    let res_tpl = admin_client.delete_records(&tpl, &opts).await.unwrap();
-    assert_eq!(res_tpl.count(), 2);
-    assert_eq!(res_tpl.elements()[0].topic(), topic);
-    assert_eq!(res_tpl.elements()[0].partition(), 0);
-    assert_eq!(res_tpl.elements()[0].offset(), Offset::Offset(2));
-    assert_eq!(res_tpl.elements()[0].error(), Ok(()));
-    assert_eq!(res_tpl.elements()[1].topic(), topic);
-    assert_eq!(res_tpl.elements()[1].partition(), 1);
-    assert_eq!(
-        res_tpl.elements()[1].error(),
-        Err(KafkaError::OffsetFetch(RDKafkaErrorCode::UnknownPartition))
-    );
-
-    // Ensure that the low water mark has advanced to 2.
-    let (lo, hi) = producer
-        .client()
-        .fetch_watermarks(&topic, 0, timeout)
-        .unwrap();
-    assert_eq!(lo, 2);
-    assert_eq!(hi, 5);
-
-    // Delete all records up to offset 5.
-    let mut tpl = TopicPartitionList::new();
-    tpl.add_partition_offset(&topic, 0, Offset::End).unwrap();
-    let res_tpl = admin_client.delete_records(&tpl, &opts).await.unwrap();
-    assert_eq!(res_tpl.count(), 1);
-    assert_eq!(res_tpl.elements()[0].topic(), topic);
-    assert_eq!(res_tpl.elements()[0].partition(), 0);
-    assert_eq!(res_tpl.elements()[0].offset(), Offset::Offset(5));
-    assert_eq!(res_tpl.elements()[0].error(), Ok(()));
-
-    // Ensure that the low water mark has advanced to 5.
-    let (lo, hi) = producer
-        .client()
-        .fetch_watermarks(&topic, 0, timeout)
-        .unwrap();
-    assert_eq!(lo, 5);
-    assert_eq!(hi, 5);
-}
-
-#[tokio::test]
-async fn test_configs() {
-    let admin_client = create_admin_client();
-    let opts = AdminOptions::new();
-    let broker = ResourceSpecifier::Broker(0);
-
-    let res = admin_client
-        .describe_configs(&[broker], &opts)
-        .await
-        .expect("describe configs failed");
-    let config = &res[0].as_ref().expect("describe configs failed");
-    let orig_val = config
-        .get("log.flush.interval.messages")
-        .expect("original config entry missing")
-        .value
-        .as_ref()
-        .expect("original value missing");
-
-    let config = AlterConfig::new(broker).set("log.flush.interval.messages", "1234");
-    let res = admin_client
-        .alter_configs(&[config], &opts)
-        .await
-        .expect("alter configs failed");
-    assert_eq!(res, &[Ok(OwnedResourceSpecifier::Broker(0))]);
-
-    let mut tries = 0;
-    loop {
-        let res = admin_client
-            .describe_configs(&[broker], &opts)
-            .await
-            .expect("describe configs failed");
-        let config = &res[0].as_ref().expect("describe configs failed");
-        let entry = config.get("log.flush.interval.messages");
-        let expected_entry = if get_broker_version() < KafkaVersion(1, 1, 0, 0) {
-            // Pre-1.1, the AlterConfig operation will silently fail, and the
-            // config will remain unchanged, which I guess is worth testing.
-            ConfigEntry {
-                name: "log.flush.interval.messages".into(),
-                value: Some(orig_val.clone()),
-                source: ConfigSource::Default,
-                is_read_only: true,
-                is_default: true,
-                is_sensitive: false,
-            }
-        } else {
-            ConfigEntry {
-                name: "log.flush.interval.messages".into(),
-                value: Some("1234".into()),
-                source: ConfigSource::DynamicBroker,
-                is_read_only: false,
-                is_default: false,
-                is_sensitive: false,
-            }
-        };
-        if entry == Some(&expected_entry) {
-            break;
-        } else if tries >= 5 {
-            panic!("{:?} != {:?}", entry, Some(&expected_entry));
-        } else {
-            tries += 1;
-            tokio::time::sleep(Duration::from_secs(1)).await;
-        }
-    }
-
-    let config = AlterConfig::new(broker).set("log.flush.interval.ms", orig_val);
-    let res = admin_client
-        .alter_configs(&[config], &opts)
-        .await
-        .expect("alter configs failed");
-    assert_eq!(res, &[Ok(OwnedResourceSpecifier::Broker(0))]);
-}
-
 #[tokio::test]
 async fn test_groups() {
     let admin_client = create_admin_client();