fix: test_consumer_groups_deletion and test_delete_unknown_group

Author: j7nw4r

tests/admin.rs

@@ -1,11 +1,11 @@
 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;
+use crate::utils::rand::{rand_test_group, rand_test_topic};
 use crate::utils::{get_broker_version, KafkaVersion};
 use backon::{BlockingRetryable, ExponentialBuilder};
 use rdkafka::admin::{
-    AdminOptions, AlterConfig, ConfigEntry, ConfigSource, NewPartitions, NewTopic,
+    AdminOptions, AlterConfig, ConfigEntry, ConfigSource, GroupResult, NewPartitions, NewTopic,
     OwnedResourceSpecifier, ResourceSpecifier, TopicReplication,
 };
 use rdkafka::error::KafkaError;
@@ -69,7 +69,7 @@ pub async fn test_topic_create_and_delete() {
 
     // Create consumer client
     let consumer_client =
-        utils::consumer::create_unsubscribed_base_consumer(&kafka_context.bootstrap_servers)
+        utils::consumer::create_unsubscribed_base_consumer(&kafka_context.bootstrap_servers, None)
             .await
             .expect("could not create consumer client");
 
@@ -276,7 +276,7 @@ pub async fn test_incorrect_replication_factors_are_ignored_when_creating_partit
 
     // Create consumer client
     let consumer_client =
-        utils::consumer::create_unsubscribed_base_consumer(&kafka_context.bootstrap_servers)
+        utils::consumer::create_unsubscribed_base_consumer(&kafka_context.bootstrap_servers, None)
             .await
             .expect("could not create consumer client");
 
@@ -361,7 +361,7 @@ pub async fn test_mixed_success_results() {
 
     // Create consumer client
     let consumer_client =
-        utils::consumer::create_unsubscribed_base_consumer(&kafka_context.bootstrap_servers)
+        utils::consumer::create_unsubscribed_base_consumer(&kafka_context.bootstrap_servers, None)
             .await
             .expect("could not create consumer client");
 
@@ -623,3 +623,43 @@ async fn test_configs() {
         .expect("alter configs failed");
     assert_eq!(res, &[Ok(OwnedResourceSpecifier::Broker(utils::BROKER_ID))]);
 }
+
+#[tokio::test]
+async fn test_groups() {
+    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();
+
+    // Verify that deleting a valid and invalid group results in a mixed result
+    // set.
+    {
+        let group_name = rand_test_group();
+        let unknown_group_name = rand_test_group();
+        create_consumer_group(&group_name).await;
+        let res = admin_client
+            .delete_groups(
+                &[&group_name, &unknown_group_name],
+                &AdminOptions::default(),
+            )
+            .await;
+        assert_eq!(
+            res,
+            Ok(vec![
+                Ok(group_name.to_string()),
+                Err((
+                    unknown_group_name.to_string(),
+                    RDKafkaErrorCode::GroupIdNotFound
+                ))
+            ])
+        );
+    }
+}

tests/consumer_groups.rs

@@ -0,0 +1,79 @@
+use crate::utils::containers::KafkaContext;
+use crate::utils::logging::init_test_logger;
+use crate::utils::rand::{rand_test_group, rand_test_topic};
+use rdkafka::admin::{AdminOptions, GroupResult, NewTopic, TopicReplication};
+use rdkafka_sys::RDKafkaErrorCode;
+
+mod utils;
+
+/// Verify that a valid group can be deleted.
+#[tokio::test]
+pub async fn test_consumer_groups_deletion() {
+    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");
+
+    // Create consumer_client
+    let group_name = rand_test_group();
+    let topic_name = rand_test_topic("test_topic");
+    let consumer_client = utils::consumer::create_unsubscribed_base_consumer(
+        &kafka_context.bootstrap_servers,
+        Some(&group_name),
+    )
+    .await
+    .expect("could not create subscribed base consumer");
+
+    admin_client
+        .create_topics(
+            &[NewTopic {
+                name: &topic_name,
+                num_partitions: 1,
+                replication: TopicReplication::Fixed(1),
+                config: vec![],
+            }],
+            &AdminOptions::default(),
+        )
+        .await
+        .expect("topic creation failed");
+
+    utils::consumer::create_consumer_group_on_topic(&consumer_client, &topic_name)
+        .await
+        .expect("could not create group");
+    let res = admin_client
+        .delete_groups(&[&group_name], &AdminOptions::default())
+        .await
+        .expect("could not delete groups");
+    assert_eq!(res, [Ok(group_name.to_string())]);
+}
+
+/// Verify that attempting to delete an unknown group returns a "group not
+/// found" error.
+#[tokio::test]
+pub async fn delete_unknown_group() {
+    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 unknown_group_name = rand_test_group();
+    let res = admin_client
+        .delete_groups(&[&unknown_group_name], &AdminOptions::default())
+        .await;
+    let expected: GroupResult = Err((unknown_group_name, RDKafkaErrorCode::NotCoordinator));
+    assert_eq!(res, Ok(vec![expected]));
+}

tests/producer.rs

@@ -39,7 +39,8 @@ pub async fn test_basic_produce() {
     }
 
     let consumer_result =
-        create_subscribed_base_consumer(&kafka_context.bootstrap_servers, &test_topic_name).await;
+        create_subscribed_base_consumer(&kafka_context.bootstrap_servers, None, &test_topic_name)
+            .await;
     let Ok(consumer) = consumer_result else {
         panic!(
             "could not create consumer: {}",

tests/test_admin.rs

@@ -119,56 +119,6 @@ fn verify_delete(topic: &str) {
     .unwrap()
 }
 
-#[tokio::test]
-async fn test_groups() {
-    let admin_client = create_admin_client();
-
-    // Verify that a valid group can be deleted.
-    {
-        let group_name = rand_test_group();
-        create_consumer_group(&group_name).await;
-        let res = admin_client
-            .delete_groups(&[&group_name], &AdminOptions::default())
-            .await;
-        assert_eq!(res, Ok(vec![Ok(group_name.to_string())]));
-    }
-
-    // Verify that attempting to delete an unknown group returns a "group not
-    // found" error.
-    {
-        let unknown_group_name = rand_test_group();
-        let res = admin_client
-            .delete_groups(&[&unknown_group_name], &AdminOptions::default())
-            .await;
-        let expected: GroupResult = Err((unknown_group_name, RDKafkaErrorCode::GroupIdNotFound));
-        assert_eq!(res, Ok(vec![expected]));
-    }
-
-    // Verify that deleting a valid and invalid group results in a mixed result
-    // set.
-    {
-        let group_name = rand_test_group();
-        let unknown_group_name = rand_test_group();
-        create_consumer_group(&group_name).await;
-        let res = admin_client
-            .delete_groups(
-                &[&group_name, &unknown_group_name],
-                &AdminOptions::default(),
-            )
-            .await;
-        assert_eq!(
-            res,
-            Ok(vec![
-                Ok(group_name.to_string()),
-                Err((
-                    unknown_group_name.to_string(),
-                    RDKafkaErrorCode::GroupIdNotFound
-                ))
-            ])
-        );
-    }
-}
-
 // Tests whether each admin operation properly reports an error if the entire
 // request fails. The original implementations failed to check this, resulting
 // in confusing situations where a failed admin request would return Ok([]).

tests/utils/consumer.rs

@@ -2,17 +2,19 @@ use crate::utils::rand::rand_test_group;
 use anyhow::{bail, Context};
 use backon::{BlockingRetryable, ExponentialBuilder};
 use rdkafka::config::FromClientConfig;
-use rdkafka::consumer::{BaseConsumer, Consumer};
+use rdkafka::consumer::{BaseConsumer, CommitMode, Consumer};
 use rdkafka::message::BorrowedMessage;
 use rdkafka::metadata::Metadata;
-use rdkafka::ClientConfig;
+use rdkafka::{ClientConfig, TopicPartitionList};
 use std::time::Duration;
 
 pub async fn create_subscribed_base_consumer(
     bootstrap_servers: &str,
+    consumer_group_option: Option<&str>,
     test_topic: &str,
 ) -> anyhow::Result<BaseConsumer> {
-    let unsubscribed_base_consumer = create_unsubscribed_base_consumer(bootstrap_servers).await?;
+    let unsubscribed_base_consumer =
+        create_unsubscribed_base_consumer(bootstrap_servers, consumer_group_option).await?;
     unsubscribed_base_consumer
         .subscribe(&[test_topic])
         .context("Failed to subscribe to topic")?;
@@ -21,9 +23,14 @@ pub async fn create_subscribed_base_consumer(
 
 pub async fn create_unsubscribed_base_consumer(
     bootstrap_servers: &str,
+    consumer_group_option: Option<&str>,
 ) -> anyhow::Result<BaseConsumer> {
+    let consumer_group_name = match consumer_group_option {
+        Some(consumer_group_name) => consumer_group_name,
+        None => &rand_test_group(),
+    };
     let mut consumer_client_config = ClientConfig::default();
-    consumer_client_config.set("group.id", rand_test_group());
+    consumer_client_config.set("group.id", consumer_group_name);
     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");
@@ -97,3 +104,27 @@ pub fn verify_topic_deleted(consumer: &BaseConsumer, topic: &str) -> anyhow::Res
     .retry(ExponentialBuilder::default().with_max_delay(Duration::from_secs(5)))
     .call()
 }
+
+pub async fn create_consumer_group_on_topic(
+    consumer_client: &BaseConsumer,
+    topic_name: &str,
+) -> anyhow::Result<()> {
+    let topic_partition_list = {
+        let mut lst = TopicPartitionList::new();
+        lst.add_partition(topic_name, 0);
+        lst
+    };
+    consumer_client
+        .assign(&topic_partition_list)
+        .context("assign topic partition list failed")?;
+    consumer_client
+        .fetch_metadata(None, Duration::from_secs(3))
+        .context("unable to fetch metadata")?;
+    (|| consumer_client.store_offset(topic_name, 0, -1))
+        .retry(ExponentialBuilder::default().with_max_delay(Duration::from_secs(5)))
+        .call()
+        .context("store offset failed")?;
+    consumer_client
+        .commit_consumer_state(CommitMode::Sync)
+        .context("commit the consumer state failed")
+}