diff --git a/core/cli/src/commands/binary_consumer_groups/get_consumer_group.rs b/core/cli/src/commands/binary_consumer_groups/get_consumer_group.rs index 39f70376a0..cb5e23420d 100644 --- a/core/cli/src/commands/binary_consumer_groups/get_consumer_group.rs +++ b/core/cli/src/commands/binary_consumer_groups/get_consumer_group.rs @@ -20,7 +20,8 @@ use anyhow::Context; use async_trait::async_trait; use comfy_table::{Table, presets::ASCII_NO_BORDERS}; use iggy_common::Client; -use iggy_common::Identifier; +use iggy_common::{Consumer, ConsumerGroupMember, Identifier, Partition}; +use std::collections::HashMap; use tracing::{Level, event}; pub struct GetConsumerGroupCmd { @@ -39,6 +40,22 @@ impl GetConsumerGroupCmd { } } +fn partition_owner(members: &[ConsumerGroupMember]) -> HashMap { + let mut partition_to_member = HashMap::new(); + for member in members { + for partition_id in &member.partitions { + partition_to_member.insert(*partition_id, member.id); + } + } + partition_to_member +} + +fn find_partition(partitions: &[Partition], partition_id: u32) -> Option<&Partition> { + partitions + .iter() + .find(|partition| partition.id == partition_id) +} + #[async_trait] impl CliCommand for GetConsumerGroupCmd { fn explain(&self) -> String { @@ -65,6 +82,89 @@ impl CliCommand for GetConsumerGroupCmd { } let consumer_group = consumer_group.unwrap(); + + let topic = client + .get_topic(&self.stream_id, &self.topic_id) + .await + .with_context(|| { + format!( + "Problem getting topic with ID: {} in stream with ID: {}", + self.topic_id, self.stream_id + ) + })? + .with_context(|| { + format!( + "Topic with ID: {} in stream with ID: {} was not found", + self.topic_id, self.stream_id + ) + })?; + + let consumer = Consumer::group(self.group_id.clone()); + let partition_to_member = partition_owner(&consumer_group.members); + + let mut partitions_table = Table::new(); + partitions_table.load_style(ASCII_NO_BORDERS); + partitions_table.set_header(vec![ + "Partition", + "Member", + "Current offset", + "Stored offset", + "Lag", + ]); + + for partition_id in 0..consumer_group.partitions_count { + let partition = find_partition(&topic.partitions, partition_id).with_context(|| { + format!( + "Partition with ID: {} in topic with ID: {} was not found", + partition_id, self.topic_id + ) + })?; + + let consumer_offset = client + .get_consumer_offset( + &consumer, + &self.stream_id, + &self.topic_id, + Some(partition_id), + ) + .await + .with_context(|| { + format!( + "Problem getting consumer offset for consumer group with ID: {} for stream with ID: {} and topic with ID: {} and partition with ID: {}", + self.group_id, self.stream_id, self.topic_id, partition_id + ) + })?; + + let member = partition_to_member + .get(&partition_id) + .map(|member_id| format!("{}", member_id)) + .unwrap_or_else(|| "-".to_string()); + + let (current_offset, stored_offset, lag) = match consumer_offset { + Some(offset_info) => { + let lag = offset_info.current_offset - offset_info.stored_offset; + ( + format!("{}", offset_info.current_offset), + format!("{}", offset_info.stored_offset), + format!("{}", lag), + ) + } + None => ( + format!("{}", partition.current_offset), + "-".to_string(), + format!("{}", partition.messages_count), + ), + }; + + partitions_table.add_row(vec![ + format!("{}", partition_id), + member, + current_offset, + stored_offset, + lag, + ]); + } + let mut table = Table::new(); table.set_header(vec!["Property", "Value"]); @@ -102,6 +202,8 @@ impl CliCommand for GetConsumerGroupCmd { table.add_row(vec!["Members", members_table.to_string().as_str()]); } + table.add_row(vec!["Partitions", partitions_table.to_string().as_str()]); + event!(target: PRINT_TARGET, Level::INFO,"{table}"); Ok(()) diff --git a/core/integration/tests/cli/consumer_group/test_consumer_group_get_command.rs b/core/integration/tests/cli/consumer_group/test_consumer_group_get_command.rs index 6ceb3c94c3..220750bca3 100644 --- a/core/integration/tests/cli/consumer_group/test_consumer_group_get_command.rs +++ b/core/integration/tests/cli/consumer_group/test_consumer_group_get_command.rs @@ -24,8 +24,10 @@ use async_trait::async_trait; use iggy::prelude::Client; use iggy::prelude::IggyExpiry; use iggy::prelude::TopicCreateOptions; +use iggy::prelude::{Consumer, IggyMessage, Partitioning}; use predicates::str::{contains, starts_with}; use serial_test::parallel; +use std::str::FromStr; struct TestConsumerGroupGetCmd { stream_id: u32, @@ -183,6 +185,284 @@ impl IggyCmdTestCase for TestConsumerGroupGetCmd { } } +struct TestConsumerGroupGetLagWithoutStoredOffsetCmd { + stream_name: String, + topic_name: String, + group_name: String, + messages_count: u32, + stream_id: u32, + topic_id: u32, + group_id: u32, +} + +impl TestConsumerGroupGetLagWithoutStoredOffsetCmd { + fn new() -> Self { + Self { + stream_name: String::from("cg-get-lag-no-offset-stream"), + topic_name: String::from("cg-get-lag-no-offset-topic"), + group_name: String::from("cg-get-lag-no-offset-group"), + messages_count: 10, + stream_id: 0, + topic_id: 0, + group_id: 0, + } + } +} + +#[async_trait] +impl IggyCmdTestCase for TestConsumerGroupGetLagWithoutStoredOffsetCmd { + async fn prepare_server_state(&mut self, client: &dyn Client) { + let stream = client.create_stream(&self.stream_name).await.unwrap(); + self.stream_id = stream.id; + + let topic = client + .create_topic( + &self.stream_id.try_into().unwrap(), + &self.topic_name, + &TopicCreateOptions { + partitions_count: Some(1), + message_expiry: Some(IggyExpiry::NeverExpire), + ..TopicCreateOptions::default() + }, + ) + .await + .unwrap(); + self.topic_id = topic.id; + + let consumer_group = client + .create_consumer_group( + &self.stream_id.try_into().unwrap(), + &self.topic_id.try_into().unwrap(), + &self.group_name, + ) + .await + .unwrap(); + self.group_id = consumer_group.id; + + let mut messages = (1..=self.messages_count) + .filter_map(|id| IggyMessage::from_str(format!("Test message {id}").as_str()).ok()) + .collect::>(); + client + .send_messages( + &self.stream_id.try_into().unwrap(), + &self.topic_id.try_into().unwrap(), + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .unwrap(); + } + + fn get_command(&self) -> IggyCmdCommand { + IggyCmdCommand::new() + .arg("consumer-group") + .arg("get") + .arg(self.stream_name.clone()) + .arg(self.topic_name.clone()) + .arg(self.group_name.clone()) + .with_env_credentials() + } + + fn verify_command(&self, command_state: Assert) { + let current_offset = self.messages_count - 1; + command_state + .success() + .stdout(contains("Stored offset")) + .stdout(contains(format!( + "{current_offset} | - | {}", + self.messages_count + ))); + } + + async fn verify_server_state(&self, client: &dyn Client) { + client + .delete_consumer_group( + &self.stream_id.try_into().unwrap(), + &self.topic_id.try_into().unwrap(), + &self.group_name.clone().try_into().unwrap(), + ) + .await + .unwrap(); + client + .delete_topic( + &self.stream_id.try_into().unwrap(), + &self.topic_id.try_into().unwrap(), + ) + .await + .unwrap(); + client + .delete_stream(&self.stream_id.try_into().unwrap()) + .await + .unwrap(); + } +} + +struct TestConsumerGroupGetLagWithStoredOffsetCmd { + stream_name: String, + topic_name: String, + group_name: String, + messages_count: u32, + stored_offset: u64, + stream_id: u32, + topic_id: u32, + group_id: u32, +} + +impl TestConsumerGroupGetLagWithStoredOffsetCmd { + fn new() -> Self { + Self { + stream_name: String::from("cg-get-lag-stored-stream"), + topic_name: String::from("cg-get-lag-stored-topic"), + group_name: String::from("cg-get-lag-stored-group"), + messages_count: 10, + stored_offset: 3, + stream_id: 0, + topic_id: 0, + group_id: 0, + } + } +} + +#[async_trait] +impl IggyCmdTestCase for TestConsumerGroupGetLagWithStoredOffsetCmd { + async fn prepare_server_state(&mut self, client: &dyn Client) { + let stream = client.create_stream(&self.stream_name).await.unwrap(); + self.stream_id = stream.id; + + let topic = client + .create_topic( + &self.stream_id.try_into().unwrap(), + &self.topic_name, + &TopicCreateOptions { + partitions_count: Some(1), + message_expiry: Some(IggyExpiry::NeverExpire), + ..TopicCreateOptions::default() + }, + ) + .await + .unwrap(); + self.topic_id = topic.id; + + let consumer_group = client + .create_consumer_group( + &self.stream_id.try_into().unwrap(), + &self.topic_id.try_into().unwrap(), + &self.group_name, + ) + .await + .unwrap(); + self.group_id = consumer_group.id; + + let join_result = client + .join_consumer_group( + &self.stream_id.try_into().unwrap(), + &self.topic_id.try_into().unwrap(), + &self.group_name.clone().try_into().unwrap(), + ) + .await; + assert!( + join_result.is_ok(), + "join_consumer_group failed: {:?}", + join_result.err() + ); + + let mut messages = (1..=self.messages_count) + .filter_map(|id| IggyMessage::from_str(format!("Test message {id}").as_str()).ok()) + .collect::>(); + client + .send_messages( + &self.stream_id.try_into().unwrap(), + &self.topic_id.try_into().unwrap(), + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .unwrap(); + + let group_consumer = Consumer::group(self.group_name.clone().try_into().unwrap()); + client + .store_consumer_offset( + &group_consumer, + &self.stream_id.try_into().unwrap(), + &self.topic_id.try_into().unwrap(), + Some(0), + self.stored_offset, + ) + .await + .unwrap(); + } + + fn get_command(&self) -> IggyCmdCommand { + IggyCmdCommand::new() + .arg("consumer-group") + .arg("get") + .arg(self.stream_name.clone()) + .arg(self.topic_name.clone()) + .arg(self.group_name.clone()) + .with_env_credentials() + } + + fn verify_command(&self, command_state: Assert) { + let current_offset = self.messages_count - 1; + let lag = current_offset as u64 - self.stored_offset; + command_state + .success() + .stdout(contains(format!( + "{current_offset} | {} | {lag}", + self.stored_offset + ))); + } + + async fn verify_server_state(&self, client: &dyn Client) { + let _ = client + .leave_consumer_group( + &self.stream_id.try_into().unwrap(), + &self.topic_id.try_into().unwrap(), + &self.group_name.clone().try_into().unwrap(), + ) + .await; + client + .delete_consumer_group( + &self.stream_id.try_into().unwrap(), + &self.topic_id.try_into().unwrap(), + &self.group_name.clone().try_into().unwrap(), + ) + .await + .unwrap(); + client + .delete_topic( + &self.stream_id.try_into().unwrap(), + &self.topic_id.try_into().unwrap(), + ) + .await + .unwrap(); + client + .delete_stream(&self.stream_id.try_into().unwrap()) + .await + .unwrap(); + } +} + +#[tokio::test] +#[parallel] +pub async fn should_show_lag_without_stored_offset() { + let mut iggy_cmd_test = IggyCmdTest::default(); + iggy_cmd_test.setup().await; + iggy_cmd_test + .execute_test(TestConsumerGroupGetLagWithoutStoredOffsetCmd::new()) + .await; +} + +#[tokio::test] +#[parallel] +pub async fn should_show_lag_with_stored_offset() { + let mut iggy_cmd_test = IggyCmdTest::default(); + iggy_cmd_test.setup().await; + iggy_cmd_test + .execute_test(TestConsumerGroupGetLagWithStoredOffsetCmd::new()) + .await; +} + #[tokio::test] #[parallel] pub async fn should_be_successful() {