This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 0d4fa738a feat(cli): display topic partitions and options in topic get
(#4558)
0d4fa738a is described below
commit 0d4fa738a2e7b08c92738569ae64f56180afecca
Author: SteveParadox <[email protected]>
AuthorDate: Wed Oct 21 08:48:46 2026 +0100
feat(cli): display topic partitions and options in topic get (#4558)
---
core/cli/src/commands/binary_topics/get_topic.rs | 298 +++++++++++++++---
.../tests/cli/topic/test_topic_get_command.rs | 347 +++++++++++++++++++--
2 files changed, 586 insertions(+), 59 deletions(-)
diff --git a/core/cli/src/commands/binary_topics/get_topic.rs
b/core/cli/src/commands/binary_topics/get_topic.rs
index 5eb5df546..fe2d73bf9 100644
--- a/core/cli/src/commands/binary_topics/get_topic.rs
+++ b/core/cli/src/commands/binary_topics/get_topic.rs
@@ -18,10 +18,11 @@
use crate::commands::cli_command::{CliCommand, PRINT_TARGET};
use anyhow::Context;
use async_trait::async_trait;
-use comfy_table::Table;
+use comfy_table::{Table, presets::ASCII_NO_BORDERS};
use iggy_common::Client;
use iggy_common::Identifier;
use iggy_common::IggyExpiry;
+use iggy_common::TopicDetails;
use tracing::{Level, event};
pub struct GetTopicCmd {
@@ -64,44 +65,271 @@ impl CliCommand for GetTopicCmd {
}
let topic = topic.unwrap();
- let mut table = Table::new();
+ let table = topic_table(&topic);
- table.set_header(vec!["Property", "Value"]);
- table.add_row(vec!["Topic id", format!("{}", topic.id).as_str()]);
- table.add_row(vec![
+ event!(target: PRINT_TARGET, Level::INFO,"{table}");
+
+ Ok(())
+ }
+}
+
+fn topic_table(topic: &TopicDetails) -> Table {
+ let mut table = Table::new();
+
+ table.set_header(vec!["Property", "Value"]);
+ table.add_row(vec!["Topic id", format!("{}", topic.id).as_str()]);
+ table.add_row(vec![
+ "Created",
+ topic.created_at.to_utc_string("%Y-%m-%d %H:%M:%S").as_str(),
+ ]);
+ table.add_row(vec!["Topic name", topic.name.as_str()]);
+ table.add_row(vec!["Topic size", format!("{}", topic.size).as_str()]);
+ table.add_row(vec![
+ "Compression",
+ topic.compression_algorithm.to_string().as_str(),
+ ]);
+ table.add_row(vec![
+ "Message expiry",
+ match topic.message_expiry {
+ IggyExpiry::NeverExpire => String::from("unlimited"),
+ IggyExpiry::ServerDefault => String::from("server_default"),
+ IggyExpiry::ExpireDuration(value) => format!("{value}"),
+ }
+ .as_str(),
+ ]);
+ table.add_row(vec![
+ "Max topic size",
+ format!("{}", topic.max_topic_size).as_str(),
+ ]);
+ table.add_row(vec![
+ "Topic message count",
+ format!("{}", topic.messages_count).as_str(),
+ ]);
+ table.add_row(vec![
+ "Partitions count",
+ format!("{}", topic.partitions_count).as_str(),
+ ]);
+
+ if !topic.partitions.is_empty() {
+ let mut partitions_table = Table::new();
+ partitions_table.load_style(ASCII_NO_BORDERS);
+ partitions_table.set_header(vec![
+ "Partition id",
"Created",
- topic.created_at.to_utc_string("%Y-%m-%d %H:%M:%S").as_str(),
- ]);
- table.add_row(vec!["Topic name", topic.name.as_str()]);
- table.add_row(vec!["Topic size", format!("{}", topic.size).as_str()]);
- table.add_row(vec![
- "Compression",
- topic.compression_algorithm.to_string().as_str(),
- ]);
- table.add_row(vec![
- "Message expiry",
- match topic.message_expiry {
- IggyExpiry::NeverExpire => String::from("unlimited"),
- IggyExpiry::ServerDefault => String::from("server_default"),
- IggyExpiry::ExpireDuration(value) => format!("{value}"),
- }
- .as_str(),
- ]);
- table.add_row(vec![
- "Max topic size",
- format!("{}", topic.max_topic_size).as_str(),
- ]);
- table.add_row(vec![
- "Topic message count",
- format!("{}", topic.messages_count).as_str(),
- ]);
- table.add_row(vec![
- "Partitions count",
- format!("{}", topic.partitions_count).as_str(),
+ "Segments count",
+ "Current offset",
+ "Size",
+ "Messages count",
]);
- event!(target: PRINT_TARGET, Level::INFO,"{table}");
+ for partition in &topic.partitions {
+ partitions_table.add_row(vec![
+ partition.id.to_string(),
+ partition.created_at.to_utc_string("%Y-%m-%d %H:%M:%S"),
+ partition.segments_count.to_string(),
+ partition.current_offset.to_string(),
+ partition.size.to_string(),
+ partition.messages_count.to_string(),
+ ]);
+ }
+ table.add_row(vec!["Partitions",
partitions_table.to_string().as_str()]);
+ }
- Ok(())
+ if !topic.options.is_empty() {
+ let mut options_table = Table::new();
+ options_table.load_style(ASCII_NO_BORDERS);
+ options_table.set_header(vec!["Option", "Value", "Source"]);
+ for (key, option) in &topic.options {
+ options_table.add_row(vec![
+ String::from_utf8_lossy(key.as_bytes()).into_owned(),
+ option.value.to_string_value(),
+ if option.explicit {
+ "explicit"
+ } else {
+ "default"
+ }
+ .to_owned(),
+ ]);
+ }
+ table.add_row(vec!["Options", options_table.to_string().as_str()]);
+ }
+
+ table
+}
+
+#[cfg(test)]
+mod tests {
+ use super::topic_table;
+ use iggy_common::{
+ CompressionAlgorithm, HeaderKey, HeaderValue, IggyByteSize,
IggyExpiry, IggyTimestamp,
+ MaxTopicSize, OptionValue, Partition, PartitionContext,
ResourceOptions, TopicDetails,
+ };
+
+ fn topic() -> TopicDetails {
+ TopicDetails {
+ id: 7,
+ created_at: IggyTimestamp::from(0_u64),
+ name: "test-topic".to_owned(),
+ size: IggyByteSize::from(0_u64),
+ message_expiry: IggyExpiry::NeverExpire,
+ compression_algorithm: CompressionAlgorithm::None,
+ max_topic_size: MaxTopicSize::Unlimited,
+ messages_count: 0,
+ partitions_count: 0,
+ partitions: Vec::new(),
+ options: ResourceOptions::default(),
+ }
+ }
+
+ #[test]
+ fn empty_collections_keep_topic_properties_without_nested_tables() {
+ let output = topic_table(&topic()).to_string();
+ assert!(output.contains("Topic name | test-topic"));
+ assert!(output.contains("Partitions count | 0"));
+ assert!(!output.contains("Partition id"));
+ assert!(!output.contains("Segments count"));
+ assert!(!output.contains("Options"));
+ }
+
+ #[test]
+ fn partition_rows_keep_distinct_fields_and_large_offsets() {
+ let mut topic = topic();
+ topic.partitions_count = 2;
+ topic.partitions = vec![
+ Partition {
+ id: 9,
+ created_at: IggyTimestamp::from(0_u64),
+ segments_count: 3,
+ current_offset: u64::MAX,
+ size: IggyByteSize::from(1230_u64),
+ messages_count: 5,
+ context: PartitionContext::default(),
+ },
+ Partition {
+ id: 2,
+ created_at: IggyTimestamp::from(1_000_000_u64),
+ segments_count: 7,
+ current_offset: 11,
+ size: IggyByteSize::from(0_u64),
+ messages_count: 13,
+ context: PartitionContext::default(),
+ },
+ ];
+ let output = topic_table(&topic).to_string();
+ let rows = output
+ .lines()
+ .map(|line| line.split_whitespace().collect::<Vec<_>>().join(" "))
+ .collect::<Vec<_>>();
+ assert!(
+ rows.contains(&format!(
+ "| | 9 | 1970-01-01 00:00:00 | 3 | {} | 1.23 KB | 5 |",
+ u64::MAX
+ )),
+ "{output}"
+ );
+ assert!(rows.contains(&"| | 2 | 1970-01-01 00:00:01 | 7 | 11 | 0 B |
13 |".to_owned()));
+ }
+
+ #[test]
+ fn options_keep_typed_values_provenance_and_unknown_keys() {
+ let mut topic = topic();
+ let long_value = "long value / café: ".repeat(10);
+ topic.options = ResourceOptions::from([
+ (
+ HeaderKey::try_from("future_option").unwrap(),
+
OptionValue::explicit(HeaderValue::try_from(long_value.as_str()).unwrap()),
+ ),
+ (
+ HeaderKey::try_from("preallocate_segments").unwrap(),
+ OptionValue::derived(HeaderValue::from(false)),
+ ),
+ (
+ HeaderKey::try_from("messages_required_to_save").unwrap(),
+ OptionValue::explicit(HeaderValue::from(17_u32)),
+ ),
+ ]);
+ let output = topic_table(&topic).to_string();
+ let rows = output
+ .lines()
+ .map(|line| line.split_whitespace().collect::<Vec<_>>().join(" "))
+ .collect::<Vec<_>>();
+ assert!(
+ rows.contains(&"| Options | Option | Value | Source |".to_owned()),
+ "{output}"
+ );
+ assert!(rows.contains(&format!(
+ "| | future_option | {} | explicit |",
+ long_value.trim()
+ )));
+ assert!(rows.contains(&"| | preallocate_segments | false | default
|".to_owned()));
+ assert!(rows.contains(&"| | messages_required_to_save | 17 | explicit
|".to_owned()));
+ let option_lines = output
+ .lines()
+ .filter(|line| {
+ line.contains("future_option")
+ || line.contains("preallocate_segments")
+ || line.contains("messages_required_to_save")
+ })
+ .collect::<Vec<_>>();
+ assert_eq!(option_lines.len(), 3);
+ assert!(output.contains(&long_value));
+ assert!(option_lines.iter().all(|line| {
+ line.chars().filter(|&character| character == '|').count() == 5
+ && line.chars().count() == option_lines[0].chars().count()
+ }));
+ }
+
+ #[test]
+ fn resolved_options_keep_full_width_values_and_sources() {
+ let mut fixture = topic();
+ fixture.options = ResourceOptions::from([
+ (
+ HeaderKey::try_from("max_topic_size").unwrap(),
+ OptionValue::derived(HeaderValue::from(u64::MAX)),
+ ),
+ (
+
HeaderKey::try_from("size_of_messages_required_to_save").unwrap(),
+ OptionValue::derived(HeaderValue::from(1_048_576_u64)),
+ ),
+ (
+ HeaderKey::try_from("messages_required_to_save").unwrap(),
+ OptionValue::derived(HeaderValue::from(1024_u32)),
+ ),
+ (
+ HeaderKey::try_from("preallocate_segments").unwrap(),
+ OptionValue::explicit(HeaderValue::from(false)),
+ ),
+ (
+ HeaderKey::try_from("durability").unwrap(),
+
OptionValue::explicit(HeaderValue::try_from("persisted").unwrap()),
+ ),
+ (
+ HeaderKey::try_from("consumer_offset_durability").unwrap(),
+
OptionValue::explicit(HeaderValue::try_from("replicated").unwrap()),
+ ),
+ ]);
+ let output = topic_table(&fixture).to_string();
+ for expected in [
+ "Option | Value",
+ "preallocate_segments | false |
explicit",
+ "messages_required_to_save | 1024 |
default",
+ "| durability | persisted |
explicit",
+ "consumer_offset_durability | replicated |
explicit",
+ ] {
+ assert!(output.contains(expected), "missing {expected}: {output}");
+ }
+ let rows = output
+ .lines()
+ .map(|line| line.split_whitespace().collect::<Vec<_>>().join(" "))
+ .collect::<Vec<_>>();
+ for expected in [
+ "| | max_topic_size | 18446744073709551615 | default |",
+ "| | size_of_messages_required_to_save | 1048576 | default |",
+ ] {
+ assert!(
+ rows.iter().any(|row| row == expected),
+ "missing {expected}: {output}"
+ );
+ }
}
}
diff --git a/core/integration/tests/cli/topic/test_topic_get_command.rs
b/core/integration/tests/cli/topic/test_topic_get_command.rs
index 5715e9211..8cf572a2e 100644
--- a/core/integration/tests/cli/topic/test_topic_get_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_get_command.rs
@@ -17,13 +17,18 @@
use crate::cli::common::{
CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestHelpCmd,
TestStreamId,
- TestTopicId, USAGE_PREFIX,
+ TestTopicId, USAGE_PREFIX, clear_session_entry,
};
+use assert_cmd::Command;
use assert_cmd::assert::Assert;
use async_trait::async_trait;
use iggy::prelude::Client;
+use iggy::prelude::Durability;
use iggy::prelude::IggyExpiry;
-use iggy::prelude::TopicCreateOptions;
+use iggy::prelude::defaults::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME};
+use iggy::prelude::{IggyMessage, Partitioning, StreamClient, TopicClient,
TopicCreateOptions};
+use integration::harness::{TestHarnessBuilder, TestServerConfig};
+use predicates::prelude::PredicateBooleanExt;
use predicates::str::{contains, starts_with};
use serial_test::parallel;
@@ -34,27 +39,46 @@ struct TestTopicGetCmd {
topic_name: String,
using_stream_id: TestStreamId,
using_topic_id: TestTopicId,
+ partitions_count: u32,
+ durability: Durability,
+ consumer_offset_durability: Durability,
+ partition_ids: Vec<u32>,
+ populate_partitions: bool,
}
impl TestTopicGetCmd {
fn new(
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
using_stream_id: TestStreamId,
using_topic_id: TestTopicId,
) -> Self {
Self {
- stream_id,
+ stream_id: 0,
stream_name,
- topic_id,
+ topic_id: 0,
topic_name,
using_stream_id,
using_topic_id,
+ partitions_count: 1,
+ durability: Durability::Replicated,
+ consumer_offset_durability: Durability::Replicated,
+ partition_ids: Vec::new(),
+ populate_partitions: false,
}
}
+ fn with_topic_options(mut self, partitions_count: u32, durability:
Durability) -> Self {
+ self.partitions_count = partitions_count;
+ self.durability = durability;
+ self.consumer_offset_durability = match durability {
+ Durability::Persisted => Durability::Replicated,
+ Durability::Replicated => Durability::Persisted,
+ };
+ self.populate_partitions = true;
+ self
+ }
+
fn to_args(&self) -> Vec<String> {
let mut command = match self.using_stream_id {
TestStreamId::Numeric => vec![format!("{}", self.stream_id)],
@@ -73,21 +97,52 @@ impl TestTopicGetCmd {
#[async_trait]
impl IggyCmdTestCase for TestTopicGetCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client.create_stream(&self.stream_name).await;
- assert!(stream.is_ok());
+ let stream = client
+ .create_stream(&self.stream_name)
+ .await
+ .expect("stream should be created");
+ self.stream_id = stream.id;
let topic = client
.create_topic(
&self.stream_name.clone().try_into().unwrap(),
&self.topic_name,
&TopicCreateOptions {
- partitions_count: Some(1),
+ partitions_count: Some(self.partitions_count),
message_expiry: Some(IggyExpiry::NeverExpire),
+ durability: self.durability,
+ consumer_offset_durability:
self.consumer_offset_durability,
+ preallocate_segments: Some(false),
..TopicCreateOptions::default()
},
)
- .await;
- assert!(topic.is_ok());
+ .await
+ .expect("topic should be created");
+ self.topic_id = topic.id;
+
+ if self.populate_partitions {
+ for (index, partition) in topic.partitions.iter().enumerate() {
+ let mut messages = (0..=index)
+ .map(|_| "topic get test
payload".parse::<IggyMessage>().unwrap())
+ .collect::<Vec<_>>();
+ client
+ .send_messages(
+ &self.stream_id.try_into().unwrap(),
+ &self.topic_id.try_into().unwrap(),
+ &Partitioning::partition_id(partition.id),
+ &mut messages,
+ )
+ .await
+ .expect("messages should be sent");
+ }
+ }
+
+ self.partition_ids = topic
+ .partitions
+ .iter()
+ .map(|partition| partition.id)
+ .collect();
+ assert_eq!(self.partition_ids.len(), self.partitions_count as usize);
}
fn get_command(&self) -> IggyCmdCommand {
@@ -112,19 +167,84 @@ impl IggyCmdTestCase for TestTopicGetCmd {
let start_message =
format!("Executing get topic with ID: {topic_id} from stream with
ID: {stream_id}");
- command_state
+ let expected_messages_count = if self.populate_partitions {
+ (1..=u64::from(self.partitions_count)).sum::<u64>()
+ } else {
+ 0
+ };
+ let assertion = command_state
.success()
.stdout(starts_with(start_message))
- .stdout(contains("Topic id | 0"))
+ .stdout(contains(format!("Topic id | {}",
self.topic_id)))
.stdout(contains(format!(
"Topic name | {}",
self.topic_name
)))
- .stdout(contains("Topic size | 0"))
+ .stdout(contains("Topic size | "))
.stdout(contains("Message expiry | unlimited"))
.stdout(contains("Max topic size | unlimited"))
- .stdout(contains("Topic message count | 0"))
- .stdout(contains("Partitions count | 1"));
+ .stdout(contains(format!(
+ "Topic message count | {}",
+ expected_messages_count
+ )))
+ .stdout(contains(format!(
+ "Partitions count | {}",
+ self.partitions_count
+ )));
+
+ let mut assertion = assertion
+ .stdout(contains("Partition id | Created"))
+ .stdout(contains("Segments count | Current offset | Size"))
+ .stdout(contains("Messages count"))
+ .stdout(contains("Option | Value"))
+ .stdout(contains("Source"))
+ .stdout(contains(
+ "preallocate_segments | false |
explicit",
+ ))
+ .stdout(contains(
+ "messages_required_to_save | 1024 |
default",
+ ));
+ assertion = match self.durability {
+ Durability::Persisted => assertion.stdout(contains(
+ "| durability | persisted |
explicit",
+ )),
+ Durability::Replicated => assertion.stdout(contains(
+ "| durability | replicated |
explicit",
+ )),
+ };
+ assertion = match self.consumer_offset_durability {
+ Durability::Persisted => assertion.stdout(contains(
+ "consumer_offset_durability | persisted |
explicit",
+ )),
+ Durability::Replicated => assertion.stdout(contains(
+ "consumer_offset_durability | replicated |
explicit",
+ )),
+ };
+ if self.populate_partitions {
+ assertion = assertion.stdout(contains("Topic size | 0
B").not());
+ } else {
+ assertion = assertion.stdout(contains("Topic size | 0
B"));
+ }
+ let stdout = std::str::from_utf8(&assertion.get_output().stdout)
+ .expect("CLI output should be UTF-8");
+ let rows = stdout
+ .lines()
+ .map(|line| line.split_whitespace().collect::<Vec<_>>().join(" "))
+ .collect::<Vec<_>>();
+ for (index, partition_id) in self.partition_ids.iter().enumerate() {
+ let prefix = format!("| | {partition_id} |");
+ let row = rows
+ .iter()
+ .find(|line| line.starts_with(&prefix))
+ .expect("partition should have a row in CLI output");
+ if self.populate_partitions {
+ assert!(row.contains(&format!("| 1 | {index} |")), "{row}");
+ assert!(row.contains(" B |") && !row.contains("| 0 B |"),
"{row}");
+ assert!(row.ends_with(&format!("| {} |", index + 1)), "{row}");
+ } else {
+ assert!(row.contains("| 1 | 0 | 0 B | 0 |"), "{row}");
+ }
+ }
}
async fn verify_server_state(&self, client: &dyn Client) {
@@ -151,9 +271,7 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestTopicGetCmd::new(
- 0,
String::from("main"),
- 0,
String::from("sync"),
TestStreamId::Named,
TestTopicId::Named,
@@ -161,9 +279,7 @@ pub async fn should_be_successful() {
.await;
iggy_cmd_test
.execute_test(TestTopicGetCmd::new(
- 1,
String::from("customer"),
- 3,
String::from("probe"),
TestStreamId::Named,
TestTopicId::Named,
@@ -171,9 +287,7 @@ pub async fn should_be_successful() {
.await;
iggy_cmd_test
.execute_test(TestTopicGetCmd::new(
- 1,
String::from("development"),
- 2,
String::from("testing"),
TestStreamId::Named,
TestTopicId::Named,
@@ -181,9 +295,7 @@ pub async fn should_be_successful() {
.await;
iggy_cmd_test
.execute_test(TestTopicGetCmd::new(
- 6,
String::from("other"),
- 1,
String::from("topic"),
TestStreamId::Named,
TestTopicId::Named,
@@ -191,6 +303,86 @@ pub async fn should_be_successful() {
.await;
}
+#[tokio::test]
+#[parallel]
+pub async fn should_show_multiple_partitions_and_persisted_durability() {
+ let mut iggy_cmd_test = IggyCmdTest::default();
+ iggy_cmd_test.setup().await;
+ iggy_cmd_test
+ .execute_test(
+ TestTopicGetCmd::new(
+ String::from("durable-stream"),
+ String::from("multi-partition-topic"),
+ TestStreamId::Named,
+ TestTopicId::Named,
+ )
+ .with_topic_options(3, Durability::Persisted),
+ )
+ .await;
+}
+
+#[tokio::test]
+#[parallel]
+pub async fn should_show_multiple_partitions_using_numeric_identifiers() {
+ let mut iggy_cmd_test = IggyCmdTest::default();
+ iggy_cmd_test.setup().await;
+ iggy_cmd_test
+ .execute_test(
+ TestTopicGetCmd::new(
+ String::from("numeric-stream"),
+ String::from("numeric-topic"),
+ TestStreamId::Numeric,
+ TestTopicId::Numeric,
+ )
+ .with_topic_options(3, Durability::Replicated),
+ )
+ .await;
+}
+
+struct TestMissingTopicGetCmd;
+
+#[async_trait]
+impl IggyCmdTestCase for TestMissingTopicGetCmd {
+ async fn prepare_server_state(&mut self, client: &dyn Client) {
+ client.create_stream("missing-topic-stream").await.unwrap();
+ }
+
+ fn get_command(&self) -> IggyCmdCommand {
+ IggyCmdCommand::new()
+ .args(vec![
+ "topic",
+ "get",
+ "missing-topic-stream",
+ "missing-topic",
+ ])
+ .with_env_credentials()
+ }
+
+ fn verify_command(&self, command_state: Assert) {
+ command_state
+ .success()
+ .stdout(contains(
+ "Topic with ID: missing-topic in stream missing-topic-stream
was not found",
+ ))
+ .stdout(predicates::str::contains("Partitions").not());
+ }
+
+ async fn verify_server_state(&self, client: &dyn Client) {
+ client
+ .delete_stream(&"missing-topic-stream".try_into().unwrap())
+ .await
+ .unwrap();
+ }
+}
+
+#[tokio::test]
+#[parallel]
+pub async fn should_report_missing_topic() {
+ let mut iggy_cmd_test = IggyCmdTest::default();
+ iggy_cmd_test.setup().await;
+ iggy_cmd_test.execute_test(TestMissingTopicGetCmd).await;
+}
+
#[tokio::test]
#[parallel]
pub async fn should_help_match() {
@@ -257,3 +449,110 @@ Options:
))
.await;
}
+
+#[tokio::test]
+#[parallel]
+#[ignore = "opt-in end-to-end test of the CLI persisted durability flag"]
+pub async fn should_show_persisted_durability_created_through_cli() {
+ let mut harness = TestHarnessBuilder::default()
+ .server(TestServerConfig::default())
+ .build()
+ .unwrap();
+ harness.start().await.expect("test server should start");
+ let client = harness.tcp_root_client().await.unwrap();
+ let stream = client.create_stream("cli-persisted-stream").await.unwrap();
+ let address = harness.server().raw_tcp_addr().unwrap();
+ let socket_address = harness.server().tcp_addr().unwrap();
+ clear_session_entry(socket_address);
+ let cli = || {
+ #[allow(deprecated)]
+ let mut command = Command::cargo_bin("iggy").unwrap();
+ command.env("IGGY_TOKEN", "").env("COLUMNS", "200").args([
+ "--tcp-server-address",
+ &address,
+ "--username",
+ DEFAULT_ROOT_USERNAME,
+ "--password",
+ DEFAULT_ROOT_PASSWORD,
+ ]);
+ command
+ };
+ cli()
+ .args([
+ "topic",
+ "create",
+ "cli-persisted-stream",
+ "cli-persisted-topic",
+ "3",
+ "none",
+ "--durability",
+ "persisted",
+ "--set",
+ "preallocate_segments=false",
+ ])
+ .assert()
+ .success();
+
+ let details = client
+ .get_topic(
+ &stream.id.try_into().unwrap(),
+ &"cli-persisted-topic".try_into().unwrap(),
+ )
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(details.partitions_count, 3);
+ assert_eq!(details.partitions.len(), 3);
+ let option = details
+ .options
+ .iter()
+ .find(|(key, _)| key.as_bytes() == b"durability")
+ .unwrap();
+ assert_eq!(option.1.value.to_string_value(), "persisted");
+
+ assert!(option.1.explicit);
+
+ let mut assertion = cli()
+ .args([
+ "topic",
+ "get",
+ "cli-persisted-stream",
+ "cli-persisted-topic",
+ ])
+ .assert()
+ .success()
+ .stdout(contains("Partitions count | 3"))
+ .stdout(contains("Partition id | Created"))
+ .stdout(contains("Segments count | Current offset | Size"))
+ .stdout(contains("Messages count"))
+ .stdout(contains("Source"))
+ .stdout(contains(
+ "| durability | persisted |
explicit",
+ ))
+ .stdout(contains(
+ "consumer_offset_durability | replicated |
explicit",
+ ))
+ .stdout(contains(
+ "preallocate_segments | false |
explicit",
+ ))
+ .stdout(contains(
+ "messages_required_to_save | 1024 |
default",
+ ));
+ for partition in &details.partitions {
+ assertion = assertion.stdout(contains(format!("| {:<12} |",
partition.id)));
+ }
+
+ client
+ .delete_topic(
+ &stream.id.try_into().unwrap(),
+ &details.id.try_into().unwrap(),
+ )
+ .await
+ .unwrap();
+ client
+ .delete_stream(&stream.id.try_into().unwrap())
+ .await
+ .unwrap();
+ clear_session_entry(socket_address);
+ harness.stop().await.unwrap();
+}
