github-actions[bot] commented on code in PR #4489:
URL: https://github.com/apache/iggy/pull/4489#discussion_r4322646613
##########
core/cli/src/commands/binary_consumer_groups/get_consumer_group.rs:
##########
@@ -65,6 +65,139 @@ 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());
+ if consumer_group.partitions_count > iggy_common::MAX_PARTITIONS_COUNT
{
Review Comment:
warning: The guard makes the command fail for a valid group when its topic
exceeds 1000 partitions, because `MAX_PARTITIONS_COUNT` caps one create request
and a topic can reach 1,000,000. Delete the guard, iterate `topic.partitions`,
and drop `should_reject_partitions_count_above_the_limit`.
##########
core/cli/src/commands/binary_consumer_groups/get_consumer_group.rs:
##########
@@ -65,6 +65,139 @@ 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());
+ if consumer_group.partitions_count > iggy_common::MAX_PARTITIONS_COUNT
{
+ anyhow::bail!(
+ "Consumer group with ID: {} reports partitions count {} above
the limit {}",
+ self.group_id,
+ consumer_group.partitions_count,
+ iggy_common::MAX_PARTITIONS_COUNT
+ );
+ }
+ // Size the indexes from the topic partition list. `partitions_count`
is a
+ // wire field, and a huge value must not allocate before the
missing-id check.
+ let index_len = topic.partitions.len();
+ let mut partitions_by_id: Vec<Option<&Partition>> = vec![None;
index_len];
+ for partition in &topic.partitions {
+ if let Some(slot) = partitions_by_id.get_mut(partition.id as
usize) {
+ *slot = Some(partition);
+ }
+ }
+
+ let mut owners: Vec<Option<u32>> = vec![None; index_len];
+ let mut members_table = Table::new();
+ let render_members = consumer_group.members_count > 0;
+ if render_members {
+ members_table.load_style(ASCII_NO_BORDERS);
+ members_table.set_header(vec!["Member id", "Partitions count",
"Partitions"]);
+ }
+ for member in &consumer_group.members {
+ for partition_id in &member.partitions {
+ if let Some(slot) = owners.get_mut(*partition_id as usize) {
+ *slot = Some(member.id);
+ }
+ }
+ if render_members {
+ members_table.add_row(vec![
+ format!("{}", member.id).as_str(),
+ format!("{}", member.partitions_count).as_str(),
+ member
+ .partitions
+ .iter()
+ .map(|partition_id| format!("{partition_id}"))
+ .collect::<Vec<String>>()
+ .join(", ")
+ .as_str(),
+ ]);
+ }
+ }
+
+ 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 = partitions_by_id
+ .get(partition_id as usize)
+ .copied()
+ .flatten()
+ .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 = owners
+ .get(partition_id as usize)
+ .copied()
+ .flatten()
+ .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;
Review Comment:
warning: `current_offset - stored_offset` underflows when a crash loses
acked but unflushed messages, because `seed_recovered_offset` keeps a stored
offset above the recovered commit head. In release the lag wraps to a huge
value, and in debug the command panics. Use
`offset_info.current_offset.saturating_sub(offset_info.stored_offset)`.
##########
core/integration/tests/cli/consumer_group/test_consumer_group_get_command.rs:
##########
@@ -183,6 +186,308 @@ 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,
Review Comment:
simplification: `TestConsumerGroupGetLagWithoutStoredOffsetCmd` stores
`group_id` but never reads it, so the field and its assignment are dead. Remove
both.
##########
core/integration/tests/cli/consumer_group/test_consumer_group_get_command.rs:
##########
@@ -183,6 +186,308 @@ 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::<Vec<_>>();
+ 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()
+ );
+
+ // The server fences a group offset write to the calling connection's
+ // membership. Join returns before that assignment is visible.
+ let deadline = Instant::now() + Duration::from_secs(10);
Review Comment:
simplification: The hand-rolled membership wait duplicates
`integration::harness::wait_for_consumer_group_assignment`, which also checks
uniqueness and balance. Call the helper instead, and widen its client parameter
to `?Sized` so `&dyn Client` fits.
##########
core/cli/src/commands/binary_consumer_groups/get_consumer_group.rs:
##########
@@ -65,6 +65,139 @@ 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());
+ if consumer_group.partitions_count > iggy_common::MAX_PARTITIONS_COUNT
{
+ anyhow::bail!(
+ "Consumer group with ID: {} reports partitions count {} above
the limit {}",
+ self.group_id,
+ consumer_group.partitions_count,
+ iggy_common::MAX_PARTITIONS_COUNT
+ );
+ }
+ // Size the indexes from the topic partition list. `partitions_count`
is a
+ // wire field, and a huge value must not allocate before the
missing-id check.
+ let index_len = topic.partitions.len();
+ let mut partitions_by_id: Vec<Option<&Partition>> = vec![None;
index_len];
+ for partition in &topic.partitions {
+ if let Some(slot) = partitions_by_id.get_mut(partition.id as
usize) {
+ *slot = Some(partition);
+ }
+ }
+
+ let mut owners: Vec<Option<u32>> = vec![None; index_len];
+ let mut members_table = Table::new();
+ let render_members = consumer_group.members_count > 0;
+ if render_members {
+ members_table.load_style(ASCII_NO_BORDERS);
+ members_table.set_header(vec!["Member id", "Partitions count",
"Partitions"]);
+ }
+ for member in &consumer_group.members {
+ for partition_id in &member.partitions {
+ if let Some(slot) = owners.get_mut(*partition_id as usize) {
+ *slot = Some(member.id);
+ }
+ }
+ if render_members {
+ members_table.add_row(vec![
+ format!("{}", member.id).as_str(),
+ format!("{}", member.partitions_count).as_str(),
+ member
+ .partitions
+ .iter()
+ .map(|partition_id| format!("{partition_id}"))
+ .collect::<Vec<String>>()
+ .join(", ")
+ .as_str(),
+ ]);
+ }
+ }
+
+ 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 {
Review Comment:
warning: The command sends one `get_consumer_offset` request per partition
and waits for each reply, so the run time grows with the partition count. Add a
batch consumer-offset read and use it once per topic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
