You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/09/21 14:30:47 UTC
(iggy) branch create_resource_option_id_removal created (now d981ac21)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch create_resource_option_id_removal
in repository https://gitbox.apache.org/repos/asf/iggy.git
at d981ac21 extra fixes
This branch includes the following new commits:
new ed1e0759 init
new d981ac21 extra fixes
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 02/02: extra fixes
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch create_resource_option_id_removal
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit d981ac214d53cad76800d0b7845392a58b99d7d7
Author: numminex <g....@gmail.com>
AuthorDate: Sun Sep 21 16:30:35 2025 +0200
extra fixes
---
core/ai/mcp/src/service/mod.rs | 9 +--
core/ai/mcp/src/service/requests.rs | 8 ---
core/bench/src/benchmarks/benchmark.rs | 3 +-
core/bench/src/benchmarks/common.rs | 5 +-
.../create_consumer_group.rs | 1 -
.../src/cli/binary_streams/create_stream.rs | 2 +-
core/cli/src/main.rs | 4 +-
.../scenarios/consumer_group_join_scenario.rs | 44 ++++++++------
...h_multiple_clients_polling_messages_scenario.rs | 8 +--
...with_single_client_polling_messages_scenario.rs | 1 -
core/integration/tests/server/scenarios/mod.rs | 4 +-
core/tools/src/data-seeder/seeder.rs | 65 ++++++++++----------
examples/rust/src/getting-started/producer/main.rs | 70 +++++++++++++++-------
examples/rust/src/multi-tenant/producer/main.rs | 2 +-
examples/rust/src/shared/system.rs | 3 +-
15 files changed, 122 insertions(+), 107 deletions(-)
diff --git a/core/ai/mcp/src/service/mod.rs b/core/ai/mcp/src/service/mod.rs
index 24d28c0c..166e839d 100644
--- a/core/ai/mcp/src/service/mod.rs
+++ b/core/ai/mcp/src/service/mod.rs
@@ -79,10 +79,10 @@ impl IggyService {
#[tool(description = "Create stream")]
pub async fn create_stream(
&self,
- Parameters(CreateStream { name, stream_id }): Parameters<CreateStream>,
+ Parameters(CreateStream { name }): Parameters<CreateStream>,
) -> Result<CallToolResult, ErrorData> {
self.permissions.ensure_create()?;
- request(self.client.create_stream(&name, stream_id).await)
+ request(self.client.create_stream(&name).await)
}
#[tool(description = "Update stream")]
@@ -146,7 +146,6 @@ impl IggyService {
partitions_count,
compression_algorithm,
replication_factor,
- topic_id,
message_expiry,
max_size,
}): Parameters<CreateTopic>,
@@ -167,7 +166,6 @@ impl IggyService {
partitions_count,
compression_algorithm,
replication_factor,
- topic_id,
message_expiry,
max_size,
)
@@ -501,13 +499,12 @@ impl IggyService {
stream_id,
topic_id,
name,
- group_id,
}): Parameters<CreateConsumerGroup>,
) -> Result<CallToolResult, ErrorData> {
self.permissions.ensure_create()?;
request(
self.client
- .create_consumer_group(&id(&stream_id)?, &id(&topic_id)?, &name, group_id)
+ .create_consumer_group(&id(&stream_id)?, &id(&topic_id)?, &name)
.await,
)
}
diff --git a/core/ai/mcp/src/service/requests.rs b/core/ai/mcp/src/service/requests.rs
index 1a8aea5c..f4ee7039 100644
--- a/core/ai/mcp/src/service/requests.rs
+++ b/core/ai/mcp/src/service/requests.rs
@@ -32,8 +32,6 @@ pub struct GetStream {
pub struct CreateStream {
#[schemars(description = "stream name (required, must be unique)")]
pub name: String,
- #[schemars(description = "stream identifier (numeric, optional)")]
- pub stream_id: Option<u32>,
}
#[derive(Debug, Deserialize, JsonSchema)]
@@ -88,9 +86,6 @@ pub struct CreateTopic {
#[schemars(description = "replication factor (optional, must be greater than 0)")]
pub replication_factor: Option<u8>,
- #[schemars(description = "topic identifier (numeric, optional)")]
- pub topic_id: Option<u32>,
-
#[schemars(description = "message expiry (optional)")]
pub message_expiry: Option<String>,
@@ -282,9 +277,6 @@ pub struct CreateConsumerGroup {
#[schemars(description = "consumer group name (required, must be unique)")]
pub name: String,
-
- #[schemars(description = "consumer group identifier (optional, number)")]
- pub group_id: Option<u32>,
}
#[derive(Debug, Deserialize, JsonSchema)]
diff --git a/core/bench/src/benchmarks/benchmark.rs b/core/bench/src/benchmarks/benchmark.rs
index 77179033..37e65b78 100644
--- a/core/bench/src/benchmarks/benchmark.rs
+++ b/core/bench/src/benchmarks/benchmark.rs
@@ -106,7 +106,7 @@ pub trait Benchmarkable: Send {
let stream_id: Identifier = stream_name.as_str().try_into()?;
if streams.iter().all(|s| s.name != stream_name) {
info!("Creating the test stream '{}'", stream_name);
- client.create_stream(&stream_name, None).await?;
+ client.create_stream(&stream_name).await?;
let topic_name = "topic-1".to_string();
let max_topic_size = self
.args()
@@ -125,7 +125,6 @@ pub trait Benchmarkable: Send {
partitions_count,
CompressionAlgorithm::default(),
None,
- None,
IggyExpiry::NeverExpire,
max_topic_size,
)
diff --git a/core/bench/src/benchmarks/common.rs b/core/bench/src/benchmarks/common.rs
index d9d19991..6b588646 100644
--- a/core/bench/src/benchmarks/common.rs
+++ b/core/bench/src/benchmarks/common.rs
@@ -84,15 +84,14 @@ pub async fn init_consumer_groups(
let topic_id: Identifier = "topic-1".try_into()?;
let consumer_group_name = format!("{CONSUMER_GROUP_NAME_PREFIX}-{consumer_group_id}");
info!(
- "Creating test consumer group: name={}, id={}, stream={}, topic={}",
- consumer_group_name, consumer_group_id, stream_name, topic_id
+ "Creating test consumer group: name={}, stream={}, topic={}",
+ consumer_group_name, stream_name, topic_id
);
match client
.create_consumer_group(
&stream_id,
&topic_id,
&consumer_group_name,
- Some(consumer_group_id),
)
.await
{
diff --git a/core/binary_protocol/src/cli/binary_consumer_groups/create_consumer_group.rs b/core/binary_protocol/src/cli/binary_consumer_groups/create_consumer_group.rs
index ed90c5b9..34e2106a 100644
--- a/core/binary_protocol/src/cli/binary_consumer_groups/create_consumer_group.rs
+++ b/core/binary_protocol/src/cli/binary_consumer_groups/create_consumer_group.rs
@@ -33,7 +33,6 @@ impl CreateConsumerGroupCmd {
stream_id: Identifier,
topic_id: Identifier,
name: String,
- _group_id: Option<u32>,
) -> Self {
Self {
create_consumer_group: CreateConsumerGroup {
diff --git a/core/binary_protocol/src/cli/binary_streams/create_stream.rs b/core/binary_protocol/src/cli/binary_streams/create_stream.rs
index ed7472c1..9fe750f8 100644
--- a/core/binary_protocol/src/cli/binary_streams/create_stream.rs
+++ b/core/binary_protocol/src/cli/binary_streams/create_stream.rs
@@ -28,7 +28,7 @@ pub struct CreateStreamCmd {
}
impl CreateStreamCmd {
- pub fn new(_stream_id: Option<u32>, name: String) -> Self {
+ pub fn new(name: String) -> Self {
Self {
create_stream: CreateStream { name },
}
diff --git a/core/cli/src/main.rs b/core/cli/src/main.rs
index d21536bb..8451b0d6 100644
--- a/core/cli/src/main.rs
+++ b/core/cli/src/main.rs
@@ -107,7 +107,7 @@ fn get_command(
match command {
Command::Stream(command) => match command {
StreamAction::Create(args) => {
- Box::new(CreateStreamCmd::new(args.stream_id, args.name.clone()))
+ Box::new(CreateStreamCmd::new(args.name.clone()))
}
StreamAction::Delete(args) => Box::new(DeleteStreamCmd::new(args.stream_id.clone())),
StreamAction::Update(args) => Box::new(UpdateStreamCmd::new(
@@ -121,7 +121,6 @@ fn get_command(
Command::Topic(command) => match command {
TopicAction::Create(args) => Box::new(CreateTopicCmd::new(
args.stream_id.clone(),
- args.topic_id,
args.partitions_count,
args.compression_algorithm,
args.name.clone(),
@@ -252,7 +251,6 @@ fn get_command(
create_args.stream_id.clone(),
create_args.topic_id.clone(),
create_args.name.clone(),
- create_args.group_id,
)),
ConsumerGroupAction::Delete(delete_args) => Box::new(DeleteConsumerGroupCmd::new(
delete_args.stream_id.clone(),
diff --git a/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs
index 1b297228..bd2a08f2 100644
--- a/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs
@@ -42,37 +42,41 @@ pub async fn run(client_factory: &dyn ClientFactory) {
login_root(&system_client).await;
// 1. Create the stream
- system_client
- .create_stream(STREAM_NAME, Some(STREAM_ID))
+ let stream = system_client
+ .create_stream(STREAM_NAME)
.await
.unwrap();
+ let stream_id = stream.id;
+
// 2. Create the topic
- system_client
+ let topic = system_client
.create_topic(
&Identifier::named(STREAM_NAME).unwrap(),
TOPIC_NAME,
PARTITIONS_COUNT,
CompressionAlgorithm::default(),
None,
- Some(TOPIC_ID),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
.await
.unwrap();
+ let topic_id = topic.id;
+
// 3. Create the consumer group
- system_client
+ let consumer_group = system_client
.create_consumer_group(
&Identifier::named(STREAM_NAME).unwrap(),
&Identifier::named(TOPIC_NAME).unwrap(),
CONSUMER_GROUP_NAME,
- Some(CONSUMER_GROUP_ID),
)
.await
.unwrap();
+ let consumer_group_id = consumer_group.id;
+
// 4. Create the users for all clients
create_user(&system_client, USERNAME_1).await;
create_user(&system_client, USERNAME_2).await;
@@ -87,10 +91,10 @@ pub async fn run(client_factory: &dyn ClientFactory) {
join_consumer_group(&client1).await;
// 5. Get client1 info and validate that it contains the single consumer group
- let client1_info = get_me_and_validate_consumer_groups(&client1).await;
+ let client1_info = get_me_and_validate_consumer_groups(&client1, stream_id, topic_id, consumer_group_id).await;
// 6. Validate that the consumer group has 1 member and this member has all partitions assigned
- let consumer_group = get_consumer_group_and_validate_members(&system_client, 1).await;
+ let consumer_group = get_consumer_group_and_validate_members(&system_client, 1, consumer_group_id).await;
let member = &consumer_group.members[0];
assert_eq!(member.id, client1_info.client_id);
assert_eq!(member.partitions_count, PARTITIONS_COUNT);
@@ -100,10 +104,10 @@ pub async fn run(client_factory: &dyn ClientFactory) {
join_consumer_group(&client2).await;
// 8. Validate that client 2 contains the single consumer group
- get_me_and_validate_consumer_groups(&client2).await;
+ get_me_and_validate_consumer_groups(&client2, stream_id, topic_id, consumer_group_id).await;
// 9. Validate that the consumer group has 2 members and partitions are distributed between them
- let consumer_group = get_consumer_group_and_validate_members(&system_client, 2).await;
+ let consumer_group = get_consumer_group_and_validate_members(&system_client, 2, consumer_group_id).await;
let member1 = &consumer_group.members[0];
let member2 = &consumer_group.members[1];
assert!(member1.partitions_count >= 1 && member1.partitions_count < PARTITIONS_COUNT);
@@ -117,10 +121,10 @@ pub async fn run(client_factory: &dyn ClientFactory) {
join_consumer_group(&client3).await;
// 11. Validate that client 3 contains the single consumer group
- get_me_and_validate_consumer_groups(&client3).await;
+ get_me_and_validate_consumer_groups(&client3, stream_id, topic_id, consumer_group_id).await;
// 12. Validate that the consumer group has 3 members and partitions are equally distributed between them
- let consumer_group = get_consumer_group_and_validate_members(&system_client, 3).await;
+ let consumer_group = get_consumer_group_and_validate_members(&system_client, 3, consumer_group_id).await;
let member1 = &consumer_group.members[0];
let member2 = &consumer_group.members[1];
let member3 = &consumer_group.members[2];
@@ -135,7 +139,12 @@ pub async fn run(client_factory: &dyn ClientFactory) {
assert_clean_system(&system_client).await;
}
-async fn get_me_and_validate_consumer_groups(client: &IggyClient) -> ClientInfoDetails {
+async fn get_me_and_validate_consumer_groups(
+ client: &IggyClient,
+ stream_id: u32,
+ topic_id: u32,
+ consumer_group_id: u32
+) -> ClientInfoDetails {
let client_info = client.get_me().await.unwrap();
assert!(client_info.client_id > 0);
@@ -143,9 +152,9 @@ async fn get_me_and_validate_consumer_groups(client: &IggyClient) -> ClientInfoD
assert_eq!(client_info.consumer_groups.len(), 1);
let consumer_group = &client_info.consumer_groups[0];
- assert_eq!(consumer_group.stream_id, STREAM_ID);
- assert_eq!(consumer_group.topic_id, TOPIC_ID);
- assert_eq!(consumer_group.group_id, CONSUMER_GROUP_ID);
+ assert_eq!(consumer_group.stream_id, stream_id);
+ assert_eq!(consumer_group.topic_id, topic_id);
+ assert_eq!(consumer_group.group_id, consumer_group_id);
client_info
}
@@ -153,6 +162,7 @@ async fn get_me_and_validate_consumer_groups(client: &IggyClient) -> ClientInfoD
async fn get_consumer_group_and_validate_members(
client: &IggyClient,
members_count: u32,
+ consumer_group_id: u32,
) -> ConsumerGroupDetails {
let consumer_group = client
.get_consumer_group(
@@ -164,7 +174,7 @@ async fn get_consumer_group_and_validate_members(
.unwrap()
.expect("Failed to get consumer group");
- assert_eq!(consumer_group.id, CONSUMER_GROUP_ID);
+ assert_eq!(consumer_group.id, consumer_group_id);
assert_eq!(consumer_group.name, CONSUMER_GROUP_NAME);
assert_eq!(consumer_group.partitions_count, PARTITIONS_COUNT);
assert_eq!(consumer_group.members_count, members_count);
diff --git a/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs
index 1d3bc543..f1e42c15 100644
--- a/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs
@@ -51,7 +51,7 @@ async fn init_system(
) {
// 1. Create the stream
system_client
- .create_stream(STREAM_NAME, Some(STREAM_ID))
+ .create_stream(STREAM_NAME)
.await
.unwrap();
@@ -63,7 +63,6 @@ async fn init_system(
PARTITIONS_COUNT,
CompressionAlgorithm::default(),
None,
- Some(TOPIC_ID),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -76,7 +75,6 @@ async fn init_system(
&Identifier::named(STREAM_NAME).unwrap(),
&Identifier::named(TOPIC_NAME).unwrap(),
CONSUMER_GROUP_NAME,
- Some(CONSUMER_GROUP_ID),
)
.await
.unwrap();
@@ -136,7 +134,7 @@ async fn execute_using_messages_key_key(
}
async fn poll_messages(client: &IggyClient) -> u32 {
- let consumer = Consumer::group(Identifier::numeric(CONSUMER_GROUP_ID).unwrap());
+ let consumer = Consumer::group(Identifier::named(CONSUMER_GROUP_NAME).unwrap());
let mut total_read_messages_count = 0;
for _ in 1..=PARTITIONS_COUNT * MESSAGES_COUNT {
let polled_messages = client
@@ -202,7 +200,7 @@ async fn execute_using_none_key(
}
async fn validate_message_polling(client: &IggyClient, consumer_group: &ConsumerGroupDetails) {
- let consumer = Consumer::group(Identifier::numeric(CONSUMER_GROUP_ID).unwrap());
+ let consumer = Consumer::group(Identifier::named(CONSUMER_GROUP_NAME).unwrap());
let client_info = client.get_me().await.unwrap();
let consumer_group_member = consumer_group
.members
diff --git a/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
index 4933a7dc..ff3b0aff 100644
--- a/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
@@ -78,7 +78,6 @@ async fn init_system(client: &IggyClient) {
let consumer_group_info = get_consumer_group(client).await;
let client_info = client.get_me().await.unwrap();
- assert_eq!(consumer_group_info.id, CONSUMER_GROUP_ID);
assert_eq!(consumer_group_info.members_count, 1);
assert_eq!(consumer_group_info.members.len(), 1);
diff --git a/core/integration/tests/server/scenarios/mod.rs b/core/integration/tests/server/scenarios/mod.rs
index 50029a8d..c63cc212 100644
--- a/core/integration/tests/server/scenarios/mod.rs
+++ b/core/integration/tests/server/scenarios/mod.rs
@@ -66,7 +66,7 @@ async fn join_consumer_group(client: &IggyClient) {
client
.join_consumer_group(
&Identifier::named(STREAM_NAME).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
)
.await
@@ -77,7 +77,7 @@ async fn leave_consumer_group(client: &IggyClient) {
client
.leave_consumer_group(
&Identifier::named(STREAM_NAME).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
)
.await
diff --git a/core/tools/src/data-seeder/seeder.rs b/core/tools/src/data-seeder/seeder.rs
index 8691b534..ec536317 100644
--- a/core/tools/src/data-seeder/seeder.rs
+++ b/core/tools/src/data-seeder/seeder.rs
@@ -21,36 +21,41 @@ use rand::Rng;
use std::collections::HashMap;
use std::str::FromStr;
-const PROD_STREAM_ID: u32 = 1;
-const TEST_STREAM_ID: u32 = 2;
-const DEV_STREAM_ID: u32 = 3;
+const PROD_STREAM_NAME: &str = "prod";
+const TEST_STREAM_NAME: &str = "test";
+const DEV_STREAM_NAME: &str = "dev";
pub async fn seed(client: &IggyClient) -> Result<(), IggyError> {
- create_streams(client).await?;
- create_topics(client).await?;
- send_messages(client).await?;
+ let streams = create_streams(client).await?;
+ create_topics(client, &streams).await?;
+ send_messages(client, &streams).await?;
Ok(())
}
-async fn create_streams(client: &IggyClient) -> Result<(), IggyError> {
- client.create_stream("prod", Some(PROD_STREAM_ID)).await?;
- client.create_stream("test", Some(TEST_STREAM_ID)).await?;
- client.create_stream("dev", Some(DEV_STREAM_ID)).await?;
- Ok(())
+async fn create_streams(client: &IggyClient) -> Result<Vec<(String, u32)>, IggyError> {
+ let mut streams = Vec::new();
+
+ let prod_stream = client.create_stream(PROD_STREAM_NAME).await?;
+ streams.push((PROD_STREAM_NAME.to_string(), prod_stream.id));
+
+ let test_stream = client.create_stream(TEST_STREAM_NAME).await?;
+ streams.push((TEST_STREAM_NAME.to_string(), test_stream.id));
+
+ let dev_stream = client.create_stream(DEV_STREAM_NAME).await?;
+ streams.push((DEV_STREAM_NAME.to_string(), dev_stream.id));
+
+ Ok(streams)
}
-async fn create_topics(client: &IggyClient) -> Result<(), IggyError> {
- let streams = [PROD_STREAM_ID, TEST_STREAM_ID, DEV_STREAM_ID];
- for stream_id in streams {
- let stream_id = stream_id.try_into()?;
+async fn create_topics(client: &IggyClient, streams: &[(String, u32)]) -> Result<(), IggyError> {
+ for (stream_name, _stream_id) in streams {
client
.create_topic(
- &stream_id,
+ &Identifier::named(stream_name).unwrap(),
"orders",
1,
Default::default(),
None,
- None,
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -58,12 +63,11 @@ async fn create_topics(client: &IggyClient) -> Result<(), IggyError> {
client
.create_topic(
- &stream_id,
+ &Identifier::named(stream_name).unwrap(),
"users",
2,
Default::default(),
None,
- None,
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -71,12 +75,11 @@ async fn create_topics(client: &IggyClient) -> Result<(), IggyError> {
client
.create_topic(
- &stream_id,
+ &Identifier::named(stream_name).unwrap(),
"notifications",
3,
Default::default(),
None,
- None,
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -84,12 +87,11 @@ async fn create_topics(client: &IggyClient) -> Result<(), IggyError> {
client
.create_topic(
- &stream_id,
+ &Identifier::named(stream_name).unwrap(),
"payments",
2,
Default::default(),
None,
- None,
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -97,12 +99,11 @@ async fn create_topics(client: &IggyClient) -> Result<(), IggyError> {
client
.create_topic(
- &stream_id,
+ &Identifier::named(stream_name).unwrap(),
"deliveries",
1,
Default::default(),
None,
- None,
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -111,9 +112,8 @@ async fn create_topics(client: &IggyClient) -> Result<(), IggyError> {
Ok(())
}
-async fn send_messages(client: &IggyClient) -> Result<(), IggyError> {
+async fn send_messages(client: &IggyClient, streams: &[(String, u32)]) -> Result<(), IggyError> {
let mut rng = rand::rng();
- let streams = [PROD_STREAM_ID, TEST_STREAM_ID, DEV_STREAM_ID];
let partitioning = Partitioning::balanced();
let message_batches_range = 100..=1000;
@@ -122,18 +122,19 @@ async fn send_messages(client: &IggyClient) -> Result<(), IggyError> {
let mut total_messages_sent = 0;
let mut total_batches_sent = 0;
- for (stream_idx, stream_id) in streams.iter().enumerate() {
- let stream_id_identifier = (*stream_id).try_into()?;
+ for (stream_idx, (stream_name, stream_id)) in streams.iter().enumerate() {
+ let stream_id_identifier = Identifier::named(stream_name).unwrap();
let topics = client.get_topics(&stream_id_identifier).await?;
tracing::info!(
- "Processing stream {} ({}/{})",
+ "Processing stream {} ({}) ({}/{})",
+ stream_name,
stream_id,
stream_idx + 1,
streams.len()
);
for (topic_idx, topic) in topics.iter().enumerate() {
- let topic_id = topic.id.try_into()?;
+ let topic_id_identifier = Identifier::named(&topic.name).unwrap();
let message_batches = rng.random_range(message_batches_range.clone());
tracing::info!(
@@ -183,7 +184,7 @@ async fn send_messages(client: &IggyClient) -> Result<(), IggyError> {
client
.send_messages(
&stream_id_identifier,
- &topic_id,
+ &topic_id_identifier,
&partitioning,
&mut messages,
)
diff --git a/examples/rust/src/getting-started/producer/main.rs b/examples/rust/src/getting-started/producer/main.rs
index 60e8a1f4..6114bc74 100644
--- a/examples/rust/src/getting-started/producer/main.rs
+++ b/examples/rust/src/getting-started/producer/main.rs
@@ -25,9 +25,9 @@ use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
use tracing_subscriber::{EnvFilter, Registry};
-const STREAM_ID: u32 = 1;
-const TOPIC_ID: u32 = 1;
-const PARTITION_ID: u32 = 1;
+const STREAM_NAME: &str = "sample-stream";
+const TOPIC_NAME: &str = "sample-topic";
+const PARTITION_ID: u32 = 0;
const BATCHES_LIMIT: u32 = 5;
#[tokio::main]
@@ -50,41 +50,65 @@ async fn main() -> Result<(), Box<dyn Error>> {
client
.login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
.await?;
- init_system(&client).await;
- produce_messages(&client).await
+ let (stream_id, topic_id) = init_system(&client).await;
+ produce_messages(&client, stream_id, topic_id).await
}
-async fn init_system(client: &IggyClient) {
- match client.create_stream("sample-stream", Some(STREAM_ID)).await {
- Ok(_) => info!("Stream was created."),
- Err(_) => warn!("Stream already exists and will not be created again."),
- }
+async fn init_system(client: &IggyClient) -> (u32, u32) {
+ let stream = match client.create_stream(STREAM_NAME).await {
+ Ok(stream) => {
+ info!("Stream was created.");
+ stream
+ }
+ Err(_) => {
+ warn!("Stream already exists and will not be created again.");
+ client.get_stream(&Identifier::named(STREAM_NAME).unwrap())
+ .await
+ .unwrap()
+ .expect("Failed to get stream")
+ }
+ };
- match client
+ let topic = match client
.create_topic(
- &STREAM_ID.try_into().unwrap(),
- "sample-topic",
+ &Identifier::named(STREAM_NAME).unwrap(),
+ TOPIC_NAME,
1,
CompressionAlgorithm::default(),
None,
- Some(TOPIC_ID),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
.await
{
- Ok(_) => info!("Topic was created."),
- Err(_) => warn!("Topic already exists and will not be created again."),
- }
+ Ok(topic) => {
+ info!("Topic was created.");
+ topic
+ }
+ Err(_) => {
+ warn!("Topic already exists and will not be created again.");
+ client.get_topic(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ )
+ .await
+ .unwrap()
+ .expect("Failed to get topic")
+ }
+ };
+
+ (stream.id, topic.id)
}
-async fn produce_messages(client: &dyn Client) -> Result<(), Box<dyn Error>> {
+async fn produce_messages(client: &dyn Client, stream_id: u32, topic_id: u32) -> Result<(), Box<dyn Error>> {
let duration = IggyDuration::from_str("500ms")?;
let mut interval = tokio::time::interval(duration.get_duration());
info!(
- "Messages will be sent to stream: {}, topic: {}, partition: {} with interval {}.",
- STREAM_ID,
- TOPIC_ID,
+ "Messages will be sent to stream: {} ({}), topic: {} ({}), partition: {} with interval {}.",
+ STREAM_NAME,
+ stream_id,
+ TOPIC_NAME,
+ topic_id,
PARTITION_ID,
duration.as_human_time_string()
);
@@ -109,8 +133,8 @@ async fn produce_messages(client: &dyn Client) -> Result<(), Box<dyn Error>> {
}
client
.send_messages(
- &STREAM_ID.try_into().unwrap(),
- &TOPIC_ID.try_into().unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&partitioning,
&mut messages,
)
diff --git a/examples/rust/src/multi-tenant/producer/main.rs b/examples/rust/src/multi-tenant/producer/main.rs
index 2a9682de..ec46086c 100644
--- a/examples/rust/src/multi-tenant/producer/main.rs
+++ b/examples/rust/src/multi-tenant/producer/main.rs
@@ -327,7 +327,7 @@ async fn create_stream_and_user(
username: &str,
client: &IggyClient,
) -> Result<(), IggyError> {
- let stream = client.create_stream(stream_name, None).await?;
+ let stream = client.create_stream(stream_name).await?;
info!("Created stream: {stream_name} with ID: {}", stream.id);
let mut streams_permissions = AHashMap::new();
streams_permissions.insert(
diff --git a/examples/rust/src/shared/system.rs b/examples/rust/src/shared/system.rs
index 30aa382b..8c53722d 100644
--- a/examples/rust/src/shared/system.rs
+++ b/examples/rust/src/shared/system.rs
@@ -82,7 +82,7 @@ pub async fn init_by_producer(args: &Args, client: &dyn Client) -> Result<(), Ig
}
info!("Stream does not exist, creating...");
- client.create_stream(&args.stream_id, None).await?;
+ client.create_stream(&args.stream_id).await?;
client
.create_topic(
&stream_id,
@@ -90,7 +90,6 @@ pub async fn init_by_producer(args: &Args, client: &dyn Client) -> Result<(), Ig
args.partitions_count,
CompressionAlgorithm::from_code(args.compression_algorithm)?,
None,
- None,
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
(iggy) 01/02: init
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch create_resource_option_id_removal
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit ed1e07590baa718f58152dd11f4a656ba08968bf
Author: numminex <g....@gmail.com>
AuthorDate: Sun Sep 21 15:57:58 2025 +0200
init
---
.../create_consumer_group.rs | 10 +-
.../src/cli/binary_streams/create_stream.rs | 11 +-
.../src/cli/binary_topics/create_topic.rs | 21 +-
.../client/binary_clients/consumer_group_client.rs | 1 -
.../src/client/binary_clients/stream_client.rs | 1 -
.../src/client/binary_clients/topic_client.rs | 1 -
.../src/client/binary_consumer_groups/mod.rs | 2 -
.../src/client/binary_streams/mod.rs | 2 -
.../src/client/binary_topics/mod.rs | 2 -
.../consumer_groups/create_consumer_group.rs | 41 +---
core/common/src/commands/streams/create_stream.rs | 41 +---
core/common/src/commands/topics/create_topic.rs | 61 ++----
.../test_consumer_group_create_command.rs | 3 +-
.../test_consumer_group_delete_command.rs | 4 +-
.../test_consumer_group_get_command.rs | 4 +-
.../test_consumer_group_list_command.rs | 196 +++++++----------
.../test_consumer_offset_get_command.rs | 163 ++++++--------
.../test_consumer_offset_set_command.rs | 3 +-
.../cli/message/test_message_flush_command.rs | 116 +++++-----
.../tests/cli/message/test_message_poll_command.rs | 149 +++++--------
.../message/test_message_poll_to_file_command.rs | 3 +-
.../tests/cli/message/test_message_send_command.rs | 170 +++++----------
.../message/test_message_send_from_file_command.rs | 3 +-
.../cli/partition/test_partition_create_command.rs | 94 ++++-----
.../cli/partition/test_partition_delete_command.rs | 94 ++++-----
.../tests/cli/stream/test_stream_delete_command.rs | 2 +-
.../tests/cli/stream/test_stream_get_command.rs | 2 +-
.../tests/cli/stream/test_stream_list_command.rs | 2 +-
.../tests/cli/stream/test_stream_purge_command.rs | 3 +-
.../tests/cli/stream/test_stream_update_command.rs | 2 +-
.../tests/cli/system/test_stats_command.rs | 3 +-
.../tests/cli/topic/test_topic_create_command.rs | 2 +-
.../tests/cli/topic/test_topic_delete_command.rs | 3 +-
.../tests/cli/topic/test_topic_get_command.rs | 3 +-
.../tests/cli/topic/test_topic_list_command.rs | 3 +-
.../tests/cli/topic/test_topic_purge_command.rs | 3 +-
.../tests/cli/topic/test_topic_update_command.rs | 3 +-
core/integration/tests/mcp/mod.rs | 5 +-
core/integration/tests/sdk/producer/background.rs | 30 +--
core/integration/tests/sdk/producer/mod.rs | 9 +-
.../scenarios/consumer_group_join_scenario.rs | 14 +-
...h_multiple_clients_polling_messages_scenario.rs | 30 +--
...with_single_client_polling_messages_scenario.rs | 40 ++--
.../server/scenarios/create_message_payload.rs | 17 +-
.../server/scenarios/delete_segments_scenario.rs | 31 ++-
.../tests/server/scenarios/encryption_scenario.rs | 6 +-
.../server/scenarios/message_headers_scenario.rs | 15 +-
.../server/scenarios/message_size_scenario.rs | 5 +-
core/integration/tests/server/scenarios/mod.rs | 21 +-
.../scenarios/stream_size_validation_scenario.rs | 3 +-
.../tests/server/scenarios/system_scenario.rs | 233 ++++++++++-----------
.../tests/server/scenarios/tcp_tls_scenario.rs | 19 +-
core/integration/tests/state/file.rs | 1 -
core/integration/tests/state/system.rs | 18 +-
.../binary_consumer_group_client.rs | 9 +-
.../src/client_wrappers/binary_stream_client.rs | 9 +-
.../sdk/src/client_wrappers/binary_topic_client.rs | 5 -
core/sdk/src/clients/binary_consumer_group.rs | 3 +-
core/sdk/src/clients/binary_streams.rs | 3 +-
core/sdk/src/clients/binary_topics.rs | 2 -
core/sdk/src/clients/consumer.rs | 4 +-
core/sdk/src/clients/producer.rs | 7 +-
core/sdk/src/http/binary_consumer_groups.rs | 2 -
core/sdk/src/http/binary_streams.rs | 2 -
core/sdk/src/http/binary_topics.rs | 2 -
.../src/stream_builder/build/build_stream_topic.rs | 7 +-
.../handlers/streams/create_stream_handler.rs | 5 +-
core/server/src/http/http_shard_wrapper.rs | 3 +-
core/server/src/http/streams.rs | 6 +-
core/server/src/shard/system/streams.rs | 1 -
70 files changed, 676 insertions(+), 1118 deletions(-)
diff --git a/core/binary_protocol/src/cli/binary_consumer_groups/create_consumer_group.rs b/core/binary_protocol/src/cli/binary_consumer_groups/create_consumer_group.rs
index e206b070..ed90c5b9 100644
--- a/core/binary_protocol/src/cli/binary_consumer_groups/create_consumer_group.rs
+++ b/core/binary_protocol/src/cli/binary_consumer_groups/create_consumer_group.rs
@@ -33,23 +33,19 @@ impl CreateConsumerGroupCmd {
stream_id: Identifier,
topic_id: Identifier,
name: String,
- group_id: Option<u32>,
+ _group_id: Option<u32>,
) -> Self {
Self {
create_consumer_group: CreateConsumerGroup {
stream_id,
topic_id,
name,
- group_id,
},
}
}
fn get_group_id_info(&self) -> String {
- match self.create_consumer_group.group_id {
- Some(group_id) => format!("ID: {group_id}"),
- None => "ID auto incremented".to_string(),
- }
+ "ID auto incremented".to_string()
}
}
@@ -67,7 +63,7 @@ impl CliCommand for CreateConsumerGroupCmd {
async fn execute_cmd(&mut self, client: &dyn Client) -> anyhow::Result<(), anyhow::Error> {
client
- .create_consumer_group(&self.create_consumer_group.stream_id, &self.create_consumer_group.topic_id, &self.create_consumer_group.name, self.create_consumer_group.group_id)
+ .create_consumer_group(&self.create_consumer_group.stream_id, &self.create_consumer_group.topic_id, &self.create_consumer_group.name)
.await
.with_context(|| {
format!(
diff --git a/core/binary_protocol/src/cli/binary_streams/create_stream.rs b/core/binary_protocol/src/cli/binary_streams/create_stream.rs
index 46ed1d74..ed7472c1 100644
--- a/core/binary_protocol/src/cli/binary_streams/create_stream.rs
+++ b/core/binary_protocol/src/cli/binary_streams/create_stream.rs
@@ -28,17 +28,14 @@ pub struct CreateStreamCmd {
}
impl CreateStreamCmd {
- pub fn new(stream_id: Option<u32>, name: String) -> Self {
+ pub fn new(_stream_id: Option<u32>, name: String) -> Self {
Self {
- create_stream: CreateStream { stream_id, name },
+ create_stream: CreateStream { name },
}
}
fn get_stream_id_info(&self) -> String {
- match self.create_stream.stream_id {
- Some(stream_id) => format!("ID: {stream_id}"),
- None => "ID auto incremented".to_string(),
- }
+ "ID auto incremented".to_string()
}
}
@@ -54,7 +51,7 @@ impl CliCommand for CreateStreamCmd {
async fn execute_cmd(&mut self, client: &dyn Client) -> anyhow::Result<(), anyhow::Error> {
client
- .create_stream(&self.create_stream.name, self.create_stream.stream_id)
+ .create_stream(&self.create_stream.name)
.await
.with_context(|| {
format!(
diff --git a/core/binary_protocol/src/cli/binary_topics/create_topic.rs b/core/binary_protocol/src/cli/binary_topics/create_topic.rs
index 56a82387..d605705e 100644
--- a/core/binary_protocol/src/cli/binary_topics/create_topic.rs
+++ b/core/binary_protocol/src/cli/binary_topics/create_topic.rs
@@ -36,7 +36,6 @@ impl CreateTopicCmd {
#[allow(clippy::too_many_arguments)]
pub fn new(
stream_id: Identifier,
- topic_id: Option<u32>,
partitions_count: u32,
compression_algorithm: CompressionAlgorithm,
name: String,
@@ -47,7 +46,6 @@ impl CreateTopicCmd {
Self {
create_topic: CreateTopic {
stream_id,
- topic_id,
partitions_count,
compression_algorithm,
name,
@@ -60,13 +58,6 @@ impl CreateTopicCmd {
replication_factor,
}
}
-
- fn get_topic_id_info(&self) -> String {
- match self.create_topic.topic_id {
- Some(topic_id) => format!("ID: {topic_id}"),
- None => "ID auto incremented".to_string(),
- }
- }
}
#[async_trait]
@@ -77,19 +68,18 @@ impl CliCommand for CreateTopicCmd {
async fn execute_cmd(&mut self, client: &dyn Client) -> anyhow::Result<(), anyhow::Error> {
client
- .create_topic(&self.create_topic.stream_id, &self.create_topic.name, self.create_topic.partitions_count, self.create_topic.compression_algorithm, self.create_topic.replication_factor, self.create_topic.topic_id, self.create_topic.message_expiry, self.create_topic.max_topic_size)
+ .create_topic(&self.create_topic.stream_id, &self.create_topic.name, self.create_topic.partitions_count, self.create_topic.compression_algorithm, self.create_topic.replication_factor, self.create_topic.message_expiry, self.create_topic.max_topic_size)
.await
.with_context(|| {
format!(
- "Problem creating topic (ID: {}, name: {}, partitions count: {}) in stream with ID: {}",
- self.create_topic.topic_id.unwrap_or(0), self.create_topic.name, self.create_topic.partitions_count, self.create_topic.stream_id
+ "Problem creating topic (name: {}, partitions count: {}) in stream with ID: {}",
+ self.create_topic.name, self.create_topic.partitions_count, self.create_topic.stream_id
)
})?;
event!(target: PRINT_TARGET, Level::INFO,
- "Topic with name: {}, {}, partitions count: {}, compression algorithm: {}, message expiry: {}, max topic size: {}, replication factor: {} created in stream with ID: {}",
+ "Topic with name: {}, partitions count: {}, compression algorithm: {}, message expiry: {}, max topic size: {}, replication factor: {} created in stream with ID: {}",
self.create_topic.name,
- self.get_topic_id_info(),
self.create_topic.partitions_count,
self.create_topic.compression_algorithm,
self.message_expiry,
@@ -104,7 +94,6 @@ impl CliCommand for CreateTopicCmd {
impl fmt::Display for CreateTopicCmd {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
- let topic_id = self.get_topic_id_info();
let topic_name = &self.create_topic.name;
let compression_algorithm = &self.create_topic.compression_algorithm;
let message_expiry = &self.message_expiry;
@@ -114,7 +103,7 @@ impl fmt::Display for CreateTopicCmd {
write!(
f,
- "create topic with name: {topic_name}, {topic_id}, message expiry: {message_expiry}, compression algorithm: {compression_algorithm}, \
+ "create topic with name: {topic_name}, message expiry: {message_expiry}, compression algorithm: {compression_algorithm}, \
max topic size: {max_topic_size}, replication factor: {replication_factor} in stream with ID: {stream_id}",
)
}
diff --git a/core/binary_protocol/src/client/binary_clients/consumer_group_client.rs b/core/binary_protocol/src/client/binary_clients/consumer_group_client.rs
index d379ca56..49f3b373 100644
--- a/core/binary_protocol/src/client/binary_clients/consumer_group_client.rs
+++ b/core/binary_protocol/src/client/binary_clients/consumer_group_client.rs
@@ -47,7 +47,6 @@ pub trait ConsumerGroupClient {
stream_id: &Identifier,
topic_id: &Identifier,
name: &str,
- group_id: Option<u32>,
) -> Result<ConsumerGroupDetails, IggyError>;
/// Delete a consumer group by unique ID or name for the given stream and topic by unique IDs or names.
///
diff --git a/core/binary_protocol/src/client/binary_clients/stream_client.rs b/core/binary_protocol/src/client/binary_clients/stream_client.rs
index bf4a7db4..490499dd 100644
--- a/core/binary_protocol/src/client/binary_clients/stream_client.rs
+++ b/core/binary_protocol/src/client/binary_clients/stream_client.rs
@@ -36,7 +36,6 @@ pub trait StreamClient {
async fn create_stream(
&self,
name: &str,
- stream_id: Option<u32>,
) -> Result<StreamDetails, IggyError>;
/// Update a stream by unique ID or name.
///
diff --git a/core/binary_protocol/src/client/binary_clients/topic_client.rs b/core/binary_protocol/src/client/binary_clients/topic_client.rs
index 3e89de03..bc2c0b20 100644
--- a/core/binary_protocol/src/client/binary_clients/topic_client.rs
+++ b/core/binary_protocol/src/client/binary_clients/topic_client.rs
@@ -48,7 +48,6 @@ pub trait TopicClient {
partitions_count: u32,
compression_algorithm: CompressionAlgorithm,
replication_factor: Option<u8>,
- topic_id: Option<u32>,
message_expiry: IggyExpiry,
max_topic_size: MaxTopicSize,
) -> Result<TopicDetails, IggyError>;
diff --git a/core/binary_protocol/src/client/binary_consumer_groups/mod.rs b/core/binary_protocol/src/client/binary_consumer_groups/mod.rs
index bad644d3..6495665f 100644
--- a/core/binary_protocol/src/client/binary_consumer_groups/mod.rs
+++ b/core/binary_protocol/src/client/binary_consumer_groups/mod.rs
@@ -71,7 +71,6 @@ impl<B: BinaryClient> ConsumerGroupClient for B {
stream_id: &Identifier,
topic_id: &Identifier,
name: &str,
- group_id: Option<u32>,
) -> Result<ConsumerGroupDetails, IggyError> {
fail_if_not_authenticated(self).await?;
let response = self
@@ -79,7 +78,6 @@ impl<B: BinaryClient> ConsumerGroupClient for B {
stream_id: stream_id.clone(),
topic_id: topic_id.clone(),
name: name.to_string(),
- group_id,
})
.await?;
mapper::map_consumer_group(response)
diff --git a/core/binary_protocol/src/client/binary_streams/mod.rs b/core/binary_protocol/src/client/binary_streams/mod.rs
index b48a017c..df93fcac 100644
--- a/core/binary_protocol/src/client/binary_streams/mod.rs
+++ b/core/binary_protocol/src/client/binary_streams/mod.rs
@@ -52,13 +52,11 @@ impl<B: BinaryClient> StreamClient for B {
async fn create_stream(
&self,
name: &str,
- stream_id: Option<u32>,
) -> Result<StreamDetails, IggyError> {
fail_if_not_authenticated(self).await?;
let response = self
.send_with_response(&CreateStream {
name: name.to_string(),
- stream_id,
})
.await?;
mapper::map_stream(response)
diff --git a/core/binary_protocol/src/client/binary_topics/mod.rs b/core/binary_protocol/src/client/binary_topics/mod.rs
index 5d1575b7..7ddb4205 100644
--- a/core/binary_protocol/src/client/binary_topics/mod.rs
+++ b/core/binary_protocol/src/client/binary_topics/mod.rs
@@ -67,7 +67,6 @@ impl<B: BinaryClient> TopicClient for B {
partitions_count: u32,
compression_algorithm: CompressionAlgorithm,
replication_factor: Option<u8>,
- topic_id: Option<u32>,
message_expiry: IggyExpiry,
max_topic_size: MaxTopicSize,
) -> Result<TopicDetails, IggyError> {
@@ -79,7 +78,6 @@ impl<B: BinaryClient> TopicClient for B {
partitions_count,
compression_algorithm,
replication_factor,
- topic_id,
message_expiry,
max_topic_size,
})
diff --git a/core/common/src/commands/consumer_groups/create_consumer_group.rs b/core/common/src/commands/consumer_groups/create_consumer_group.rs
index 265b0659..fb11a7c2 100644
--- a/core/common/src/commands/consumer_groups/create_consumer_group.rs
+++ b/core/common/src/commands/consumer_groups/create_consumer_group.rs
@@ -32,7 +32,6 @@ use std::str::from_utf8;
/// It has additional payload:
/// - `stream_id` - unique stream ID (numeric or name).
/// - `topic_id` - unique topic ID (numeric or name).
-/// - `group_id` - unique consumer group ID.
/// - `name` - unique consumer group name, max length is 255 characters.
#[derive(Debug, Serialize, Deserialize, PartialEq)]
pub struct CreateConsumerGroup {
@@ -42,8 +41,6 @@ pub struct CreateConsumerGroup {
/// Unique topic ID (numeric or name).
#[serde(skip)]
pub topic_id: Identifier,
- /// Unique consumer group ID (numeric), if None is provided then the server will automatically assign it.
- pub group_id: Option<u32>,
/// Unique consumer group name, max length is 255 characters.
pub name: String,
}
@@ -59,7 +56,6 @@ impl Default for CreateConsumerGroup {
CreateConsumerGroup {
stream_id: Identifier::default(),
topic_id: Identifier::default(),
- group_id: None,
name: "consumer_group_1".to_string(),
}
}
@@ -67,12 +63,6 @@ impl Default for CreateConsumerGroup {
impl Validatable<IggyError> for CreateConsumerGroup {
fn validate(&self) -> Result<(), IggyError> {
- if let Some(group_id) = self.group_id
- && group_id == 0
- {
- return Err(IggyError::InvalidConsumerGroupId);
- }
-
if self.name.is_empty() || self.name.len() > MAX_NAME_LENGTH {
return Err(IggyError::InvalidConsumerGroupName);
}
@@ -86,11 +76,10 @@ impl BytesSerializable for CreateConsumerGroup {
let stream_id_bytes = self.stream_id.to_bytes();
let topic_id_bytes = self.topic_id.to_bytes();
let mut bytes = BytesMut::with_capacity(
- 4 + stream_id_bytes.len() + topic_id_bytes.len() + self.name.len(),
+ 1 + stream_id_bytes.len() + topic_id_bytes.len() + self.name.len(),
);
bytes.put_slice(&stream_id_bytes);
bytes.put_slice(&topic_id_bytes);
- bytes.put_u32_le(self.group_id.unwrap_or(0));
#[allow(clippy::cast_possible_truncation)]
bytes.put_u8(self.name.len() as u8);
bytes.put_slice(self.name.as_bytes());
@@ -98,7 +87,7 @@ impl BytesSerializable for CreateConsumerGroup {
}
fn from_bytes(bytes: Bytes) -> Result<CreateConsumerGroup, IggyError> {
- if bytes.len() < 10 {
+ if bytes.len() < 6 {
return Err(IggyError::InvalidCommand);
}
@@ -107,20 +96,13 @@ impl BytesSerializable for CreateConsumerGroup {
position += stream_id.get_size_bytes().as_bytes_usize();
let topic_id = Identifier::from_bytes(bytes.slice(position..))?;
position += topic_id.get_size_bytes().as_bytes_usize();
- let group_id = u32::from_le_bytes(
- bytes[position..position + 4]
- .try_into()
- .map_err(|_| IggyError::InvalidNumberEncoding)?,
- );
- let group_id = if group_id == 0 { None } else { Some(group_id) };
- let name_length = bytes[position + 4];
- let name = from_utf8(&bytes[position + 5..position + 5 + name_length as usize])
+ let name_length = bytes[position];
+ let name = from_utf8(&bytes[position + 1..position + 1 + name_length as usize])
.map_err(|_| IggyError::InvalidUtf8)?
.to_string();
let command = CreateConsumerGroup {
stream_id,
topic_id,
- group_id,
name,
};
Ok(command)
@@ -131,10 +113,9 @@ impl Display for CreateConsumerGroup {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{}|{}|{}|{}",
+ "{}|{}|{}",
self.stream_id,
self.topic_id,
- self.group_id.unwrap_or(0),
self.name
)
}
@@ -149,7 +130,6 @@ mod tests {
let command = CreateConsumerGroup {
stream_id: Identifier::numeric(1).unwrap(),
topic_id: Identifier::numeric(2).unwrap(),
- group_id: Some(3),
name: "test".to_string(),
};
@@ -159,14 +139,12 @@ mod tests {
position += stream_id.get_size_bytes().as_bytes_usize();
let topic_id = Identifier::from_bytes(bytes.slice(position..)).unwrap();
position += topic_id.get_size_bytes().as_bytes_usize();
- let group_id = u32::from_le_bytes(bytes[position..position + 4].try_into().unwrap());
- let name_length = bytes[position + 4];
- let name = from_utf8(&bytes[position + 5..position + 5 + name_length as usize]).unwrap();
+ let name_length = bytes[position];
+ let name = from_utf8(&bytes[position + 1..position + 1 + name_length as usize]).unwrap();
assert!(!bytes.is_empty());
assert_eq!(stream_id, command.stream_id);
assert_eq!(topic_id, command.topic_id);
- assert_eq!(group_id, command.group_id.unwrap());
assert_eq!(name, command.name);
}
@@ -174,15 +152,13 @@ mod tests {
fn should_be_deserialized_from_bytes() {
let stream_id = Identifier::numeric(1).unwrap();
let topic_id = Identifier::numeric(2).unwrap();
- let group_id = 3u32;
let name = "test".to_string();
let stream_id_bytes = stream_id.to_bytes();
let topic_id_bytes = topic_id.to_bytes();
let mut bytes =
- BytesMut::with_capacity(4 + stream_id_bytes.len() + topic_id_bytes.len() + name.len());
+ BytesMut::with_capacity(1 + stream_id_bytes.len() + topic_id_bytes.len() + name.len());
bytes.put_slice(&stream_id_bytes);
bytes.put_slice(&topic_id_bytes);
- bytes.put_u32_le(group_id);
#[allow(clippy::cast_possible_truncation)]
bytes.put_u8(name.len() as u8);
bytes.put_slice(name.as_bytes());
@@ -192,7 +168,6 @@ mod tests {
let command = command.unwrap();
assert_eq!(command.stream_id, stream_id);
assert_eq!(command.topic_id, topic_id);
- assert_eq!(command.group_id.unwrap(), group_id);
assert_eq!(command.name, name);
}
}
diff --git a/core/common/src/commands/streams/create_stream.rs b/core/common/src/commands/streams/create_stream.rs
index c163e772..292cfcfc 100644
--- a/core/common/src/commands/streams/create_stream.rs
+++ b/core/common/src/commands/streams/create_stream.rs
@@ -28,12 +28,9 @@ use std::str::from_utf8;
/// `CreateStream` command is used to create a new stream.
/// It has additional payload:
-/// - `stream_id` - unique stream ID (numeric)
/// - `name` - unique stream name (string), max length is 255 characters.
#[derive(Debug, Serialize, Deserialize, PartialEq)]
pub struct CreateStream {
- /// Unique stream ID (numeric), if None is provided then the server will automatically assign it.
- pub stream_id: Option<u32>,
/// Unique stream name (string), max length is 255 characters.
pub name: String,
}
@@ -47,7 +44,6 @@ impl Command for CreateStream {
impl Default for CreateStream {
fn default() -> Self {
CreateStream {
- stream_id: Some(1),
name: "stream".to_string(),
}
}
@@ -55,12 +51,6 @@ impl Default for CreateStream {
impl Validatable<IggyError> for CreateStream {
fn validate(&self) -> Result<(), IggyError> {
- if let Some(stream_id) = self.stream_id
- && stream_id == 0
- {
- return Err(IggyError::InvalidStreamId);
- }
-
if self.name.is_empty() || self.name.len() > MAX_NAME_LENGTH {
return Err(IggyError::InvalidStreamName);
}
@@ -72,7 +62,6 @@ impl Validatable<IggyError> for CreateStream {
impl BytesSerializable for CreateStream {
fn to_bytes(&self) -> Bytes {
let mut bytes = BytesMut::with_capacity(5 + self.name.len());
- bytes.put_u32_le(self.stream_id.unwrap_or(0));
#[allow(clippy::cast_possible_truncation)]
bytes.put_u8(self.name.len() as u8);
bytes.put_slice(self.name.as_bytes());
@@ -80,36 +69,26 @@ impl BytesSerializable for CreateStream {
}
fn from_bytes(bytes: Bytes) -> Result<CreateStream, IggyError> {
- if bytes.len() < 6 {
+ if bytes.len() < 2 {
return Err(IggyError::InvalidCommand);
}
- let stream_id = u32::from_le_bytes(
- bytes[..4]
- .try_into()
- .map_err(|_| IggyError::InvalidNumberEncoding)?,
- );
- let stream_id = if stream_id == 0 {
- None
- } else {
- Some(stream_id)
- };
- let name_length = bytes[4];
- let name = from_utf8(&bytes[5..5 + name_length as usize])
+ let name_length = bytes[0];
+ let name = from_utf8(&bytes[1..1 + name_length as usize])
.map_err(|_| IggyError::InvalidUtf8)?
.to_string();
if name.len() != name_length as usize {
return Err(IggyError::InvalidCommand);
}
- let command = CreateStream { stream_id, name };
+ let command = CreateStream { name };
Ok(command)
}
}
impl Display for CreateStream {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- write!(f, "{}|{}", self.stream_id.unwrap_or(0), self.name)
+ write!(f, "{}", self.name)
}
}
@@ -120,26 +99,21 @@ mod tests {
#[test]
fn should_be_serialized_as_bytes() {
let command = CreateStream {
- stream_id: Some(1),
name: "test".to_string(),
};
let bytes = command.to_bytes();
- let stream_id = u32::from_le_bytes(bytes[..4].try_into().unwrap());
- let name_length = bytes[4];
- let name = from_utf8(&bytes[5..5 + name_length as usize]).unwrap();
+ let name_length = bytes[0];
+ let name = from_utf8(&bytes[1..1 + name_length as usize]).unwrap();
assert!(!bytes.is_empty());
- assert_eq!(stream_id, command.stream_id.unwrap());
assert_eq!(name, command.name);
}
#[test]
fn should_be_deserialized_from_bytes() {
- let stream_id = 1u32;
let name = "test".to_string();
let mut bytes = BytesMut::new();
- bytes.put_u32_le(stream_id);
#[allow(clippy::cast_possible_truncation)]
bytes.put_u8(name.len() as u8);
bytes.put_slice(name.as_bytes());
@@ -147,7 +121,6 @@ mod tests {
assert!(command.is_ok());
let command = command.unwrap();
- assert_eq!(command.stream_id.unwrap(), stream_id);
assert_eq!(command.name, name);
}
}
diff --git a/core/common/src/commands/topics/create_topic.rs b/core/common/src/commands/topics/create_topic.rs
index f8ac2da0..ba90084e 100644
--- a/core/common/src/commands/topics/create_topic.rs
+++ b/core/common/src/commands/topics/create_topic.rs
@@ -34,7 +34,6 @@ use std::str::from_utf8;
/// `CreateTopic` command is used to create a new topic in a stream.
/// It has additional payload:
/// - `stream_id` - unique stream ID (numeric or name).
-/// - `topic_id` - unique topic ID (numeric).
/// - `partitions_count` - number of partitions in the topic, max value is 1000.
/// - `message_expiry` - message expiry, if `NeverExpire` then messages will never expire.
/// - `max_topic_size` - maximum size of the topic, if `Unlimited` then topic size is unlimited.
@@ -46,8 +45,6 @@ pub struct CreateTopic {
/// Unique stream ID (numeric or name).
#[serde(skip)]
pub stream_id: Identifier,
- /// Unique topic ID (numeric), if None is provided then the server will automatically assign it.
- pub topic_id: Option<u32>,
/// Number of partitions in the topic, max value is 1000.
pub partitions_count: u32,
/// Compression algorithm for the topic.
@@ -73,7 +70,6 @@ impl Default for CreateTopic {
fn default() -> Self {
CreateTopic {
stream_id: Identifier::default(),
- topic_id: Some(1),
partitions_count: 1,
compression_algorithm: CompressionAlgorithm::None,
message_expiry: IggyExpiry::NeverExpire,
@@ -86,12 +82,6 @@ impl Default for CreateTopic {
impl Validatable<IggyError> for CreateTopic {
fn validate(&self) -> Result<(), IggyError> {
- if let Some(topic_id) = self.topic_id
- && topic_id == 0
- {
- return Err(IggyError::InvalidTopicId);
- }
-
if self.name.is_empty() || self.name.len() > MAX_NAME_LENGTH {
return Err(IggyError::InvalidTopicName);
}
@@ -113,9 +103,8 @@ impl Validatable<IggyError> for CreateTopic {
impl BytesSerializable for CreateTopic {
fn to_bytes(&self) -> Bytes {
let stream_id_bytes = self.stream_id.to_bytes();
- let mut bytes = BytesMut::with_capacity(23 + stream_id_bytes.len() + self.name.len());
+ let mut bytes = BytesMut::with_capacity(19 + stream_id_bytes.len() + self.name.len());
bytes.put_slice(&stream_id_bytes);
- bytes.put_u32_le(self.topic_id.unwrap_or(0));
bytes.put_u32_le(self.partitions_count);
bytes.put_u8(self.compression_algorithm.as_code());
bytes.put_u64_le(self.message_expiry.into());
@@ -131,42 +120,36 @@ impl BytesSerializable for CreateTopic {
}
fn from_bytes(bytes: Bytes) -> std::result::Result<CreateTopic, IggyError> {
- if bytes.len() < 18 {
+ if bytes.len() < 14 {
return Err(IggyError::InvalidCommand);
}
let mut position = 0;
let stream_id = Identifier::from_bytes(bytes.clone())?;
position += stream_id.get_size_bytes().as_bytes_usize();
- let topic_id = u32::from_le_bytes(
- bytes[position..position + 4]
- .try_into()
- .map_err(|_| IggyError::InvalidNumberEncoding)?,
- );
- let topic_id = if topic_id == 0 { None } else { Some(topic_id) };
let partitions_count = u32::from_le_bytes(
- bytes[position + 4..position + 8]
+ bytes[position..position + 4]
.try_into()
.map_err(|_| IggyError::InvalidNumberEncoding)?,
);
- let compression_algorithm = CompressionAlgorithm::from_code(bytes[position + 8])?;
+ let compression_algorithm = CompressionAlgorithm::from_code(bytes[position + 4])?;
let message_expiry = u64::from_le_bytes(
- bytes[position + 9..position + 17]
+ bytes[position + 5..position + 13]
.try_into()
.map_err(|_| IggyError::InvalidNumberEncoding)?,
);
let message_expiry: IggyExpiry = message_expiry.into();
let max_topic_size = u64::from_le_bytes(
- bytes[position + 17..position + 25]
+ bytes[position + 13..position + 21]
.try_into()
.map_err(|_| IggyError::InvalidNumberEncoding)?,
);
let max_topic_size: MaxTopicSize = max_topic_size.into();
- let replication_factor = match bytes[position + 25] {
+ let replication_factor = match bytes[position + 21] {
0 => None,
factor => Some(factor),
};
- let name_length = bytes[position + 26];
- let name = from_utf8(&bytes[position + 27..(position + 27 + name_length as usize)])
+ let name_length = bytes[position + 22];
+ let name = from_utf8(&bytes[position + 23..(position + 23 + name_length as usize)])
.map_err(|_| IggyError::InvalidUtf8)?
.to_string();
if name.len() != name_length as usize {
@@ -174,7 +157,6 @@ impl BytesSerializable for CreateTopic {
}
let command = CreateTopic {
stream_id,
- topic_id,
partitions_count,
compression_algorithm,
message_expiry,
@@ -190,9 +172,8 @@ impl Display for CreateTopic {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{}|{}|{}|{}|{}|{}|{}",
+ "{}|{}|{}|{}|{}|{}",
self.stream_id,
- self.topic_id.unwrap_or(0),
self.partitions_count,
self.message_expiry,
self.max_topic_size,
@@ -211,7 +192,6 @@ mod tests {
fn should_be_serialized_as_bytes() {
let command = CreateTopic {
stream_id: Identifier::numeric(1).unwrap(),
- topic_id: Some(2),
partitions_count: 3,
message_expiry: IggyExpiry::NeverExpire,
compression_algorithm: CompressionAlgorithm::None,
@@ -223,25 +203,23 @@ mod tests {
let mut position = 0;
let stream_id = Identifier::from_bytes(bytes.clone()).unwrap();
position += stream_id.get_size_bytes().as_bytes_usize();
- let topic_id = u32::from_le_bytes(bytes[position..position + 4].try_into().unwrap());
let partitions_count =
- u32::from_le_bytes(bytes[position + 4..position + 8].try_into().unwrap());
- let compression_algorithm = CompressionAlgorithm::from_code(bytes[position + 8]).unwrap();
+ u32::from_le_bytes(bytes[position..position + 4].try_into().unwrap());
+ let compression_algorithm = CompressionAlgorithm::from_code(bytes[position + 4]).unwrap();
let message_expiry =
- u64::from_le_bytes(bytes[position + 9..position + 17].try_into().unwrap());
+ u64::from_le_bytes(bytes[position + 5..position + 13].try_into().unwrap());
let message_expiry: IggyExpiry = message_expiry.into();
let max_topic_size =
- u64::from_le_bytes(bytes[position + 17..position + 25].try_into().unwrap());
+ u64::from_le_bytes(bytes[position + 13..position + 21].try_into().unwrap());
let max_topic_size: MaxTopicSize = max_topic_size.into();
- let replication_factor = bytes[position + 25];
- let name_length = bytes[position + 26];
- let name = from_utf8(&bytes[position + 27..(position + 27 + name_length as usize)])
+ let replication_factor = bytes[position + 21];
+ let name_length = bytes[position + 22];
+ let name = from_utf8(&bytes[position + 23..(position + 23 + name_length as usize)])
.unwrap()
.to_string();
assert!(!bytes.is_empty());
assert_eq!(stream_id, command.stream_id);
- assert_eq!(topic_id, command.topic_id.unwrap());
assert_eq!(partitions_count, command.partitions_count);
assert_eq!(compression_algorithm, command.compression_algorithm);
assert_eq!(message_expiry, command.message_expiry);
@@ -254,7 +232,6 @@ mod tests {
#[test]
fn should_be_deserialized_from_bytes() {
let stream_id = Identifier::numeric(1).unwrap();
- let topic_id = 2u32;
let partitions_count = 3u32;
let compression_algorithm = CompressionAlgorithm::None;
let name = "test".to_string();
@@ -262,9 +239,8 @@ mod tests {
let max_topic_size = MaxTopicSize::ServerDefault;
let replication_factor = 1;
let stream_id_bytes = stream_id.to_bytes();
- let mut bytes = BytesMut::with_capacity(14 + stream_id_bytes.len() + name.len());
+ let mut bytes = BytesMut::with_capacity(10 + stream_id_bytes.len() + name.len());
bytes.put_slice(&stream_id_bytes);
- bytes.put_u32_le(topic_id);
bytes.put_u32_le(partitions_count);
bytes.put_u8(compression_algorithm.as_code());
bytes.put_u64_le(message_expiry.into());
@@ -279,7 +255,6 @@ mod tests {
let command = command.unwrap();
assert_eq!(command.stream_id, stream_id);
- assert_eq!(command.topic_id.unwrap(), topic_id);
assert_eq!(command.name, name);
assert_eq!(command.partitions_count, partitions_count);
assert_eq!(command.compression_algorithm, compression_algorithm);
diff --git a/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs b/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
index 213831f0..56bcef2c 100644
--- a/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
+++ b/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
@@ -87,7 +87,7 @@ impl TestConsumerGroupCreateCmd {
impl IggyCmdTestCase for TestConsumerGroupCreateCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client
- .create_stream(&self.stream_name, self.stream_id.into())
+ .create_stream(&self.stream_name)
.await;
assert!(stream.is_ok());
@@ -98,7 +98,6 @@ impl IggyCmdTestCase for TestConsumerGroupCreateCmd {
1,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
diff --git a/core/integration/tests/cli/consumer_group/test_consumer_group_delete_command.rs b/core/integration/tests/cli/consumer_group/test_consumer_group_delete_command.rs
index a6f95044..df9c5b09 100644
--- a/core/integration/tests/cli/consumer_group/test_consumer_group_delete_command.rs
+++ b/core/integration/tests/cli/consumer_group/test_consumer_group_delete_command.rs
@@ -90,7 +90,7 @@ impl TestConsumerGroupDeleteCmd {
impl IggyCmdTestCase for TestConsumerGroupDeleteCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client
- .create_stream(&self.stream_name, Some(self.stream_id))
+ .create_stream(&self.stream_name)
.await;
assert!(stream.is_ok());
@@ -101,7 +101,6 @@ impl IggyCmdTestCase for TestConsumerGroupDeleteCmd {
1,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -113,7 +112,6 @@ impl IggyCmdTestCase for TestConsumerGroupDeleteCmd {
&self.stream_id.try_into().unwrap(),
&self.topic_id.try_into().unwrap(),
&self.group_name,
- Some(self.group_id),
)
.await;
assert!(consumer_group.is_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 85a58c12..5d0ebdd9 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
@@ -90,7 +90,7 @@ impl TestConsumerGroupGetCmd {
impl IggyCmdTestCase for TestConsumerGroupGetCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client
- .create_stream(&self.stream_name, Some(self.stream_id))
+ .create_stream(&self.stream_name)
.await;
assert!(stream.is_ok());
@@ -101,7 +101,6 @@ impl IggyCmdTestCase for TestConsumerGroupGetCmd {
0,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -113,7 +112,6 @@ impl IggyCmdTestCase for TestConsumerGroupGetCmd {
&self.stream_id.try_into().unwrap(),
&self.topic_id.try_into().unwrap(),
&self.group_name,
- self.group_id.into(),
)
.await;
assert!(consumer_group.is_ok());
diff --git a/core/integration/tests/cli/consumer_group/test_consumer_group_list_command.rs b/core/integration/tests/cli/consumer_group/test_consumer_group_list_command.rs
index 88712d14..7f66937a 100644
--- a/core/integration/tests/cli/consumer_group/test_consumer_group_list_command.rs
+++ b/core/integration/tests/cli/consumer_group/test_consumer_group_list_command.rs
@@ -18,7 +18,7 @@
use crate::cli::common::{
CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, OutputFormat, TestHelpCmd,
- TestStreamId, TestTopicId, USAGE_PREFIX,
+ USAGE_PREFIX,
};
use assert_cmd::assert::Assert;
use async_trait::async_trait;
@@ -29,53 +29,50 @@ use predicates::str::{contains, starts_with};
use serial_test::parallel;
struct TestConsumerGroupListCmd {
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
- consumer_group_id: u32,
consumer_group_name: String,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
output: OutputFormat,
+ // These will be populated after creating the resources
+ actual_stream_id: Option<u32>,
+ actual_topic_id: Option<u32>,
+ actual_consumer_group_id: Option<u32>,
}
impl TestConsumerGroupListCmd {
- #[allow(clippy::too_many_arguments)]
fn new(
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
- consumer_group_id: u32,
consumer_group_name: String,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
output: OutputFormat,
) -> Self {
Self {
- stream_id,
stream_name,
- topic_id,
topic_name,
- consumer_group_id,
consumer_group_name,
- using_stream_id,
- using_topic_id,
output,
+ actual_stream_id: None,
+ actual_topic_id: None,
+ actual_consumer_group_id: None,
}
}
fn to_args(&self) -> Vec<String> {
- let mut command = match self.using_stream_id {
- TestStreamId::Numeric => vec![format!("{}", self.stream_id)],
- TestStreamId::Named => vec![self.stream_name.clone()],
- };
+ let mut command = vec![];
+
+ // Use actual stream ID if available, otherwise use stream name as fallback
+ if let Some(stream_id) = self.actual_stream_id {
+ command.push(format!("{}", stream_id));
+ } else {
+ command.push(self.stream_name.clone());
+ }
- command.push(match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
- });
+ // Use actual topic ID if available, otherwise use topic name as fallback
+ if let Some(topic_id) = self.actual_topic_id {
+ command.push(format!("{}", topic_id));
+ } else {
+ command.push(self.topic_name.clone());
+ }
command.extend(self.output.to_args().into_iter().map(String::from));
@@ -86,34 +83,38 @@ impl TestConsumerGroupListCmd {
#[async_trait]
impl IggyCmdTestCase for TestConsumerGroupListCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
+ // Create stream and capture its actual ID
let stream = client
- .create_stream(&self.stream_name, self.stream_id.into())
- .await;
- assert!(stream.is_ok());
+ .create_stream(&self.stream_name)
+ .await
+ .expect("Failed to create stream");
+ self.actual_stream_id = Some(stream.id);
+ // Create topic and capture its actual ID
let topic = client
.create_topic(
- &self.stream_id.try_into().unwrap(),
+ &stream.id.try_into().unwrap(),
&self.topic_name,
1,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
- .await;
- assert!(topic.is_ok());
+ .await
+ .expect("Failed to create topic");
+ self.actual_topic_id = Some(topic.id);
+ // Create consumer group and capture its actual ID
let consumer_group = client
.create_consumer_group(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &stream.id.try_into().unwrap(),
+ &topic.id.try_into().unwrap(),
&self.consumer_group_name,
- self.consumer_group_id.into(),
)
- .await;
- assert!(consumer_group.is_ok());
+ .await
+ .expect("Failed to create consumer group");
+ self.actual_consumer_group_id = Some(consumer_group.id);
}
fn get_command(&self) -> IggyCmdCommand {
@@ -125,14 +126,16 @@ impl IggyCmdTestCase for TestConsumerGroupListCmd {
}
fn verify_command(&self, command_state: Assert) {
- let stream_id = match self.using_stream_id {
- TestStreamId::Numeric => format!("{}", self.stream_id),
- TestStreamId::Named => self.stream_name.clone(),
+ let stream_id = if let Some(stream_id) = self.actual_stream_id {
+ format!("{}", stream_id)
+ } else {
+ self.stream_name.clone()
};
- let topic_id = match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
+ let topic_id = if let Some(topic_id) = self.actual_topic_id {
+ format!("{}", topic_id)
+ } else {
+ self.topic_name.clone()
};
let start_message = format!(
@@ -147,27 +150,34 @@ impl IggyCmdTestCase for TestConsumerGroupListCmd {
}
async fn verify_server_state(&self, client: &dyn Client) {
- let consumer_group = client
- .delete_consumer_group(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
- &self.consumer_group_id.try_into().unwrap(),
- )
- .await;
- assert!(consumer_group.is_ok());
+ // Use the actual IDs that were generated during resource creation
+ if let Some(consumer_group_id) = self.actual_consumer_group_id {
+ let consumer_group = client
+ .delete_consumer_group(
+ &self.actual_stream_id.unwrap().try_into().unwrap(),
+ &self.actual_topic_id.unwrap().try_into().unwrap(),
+ &consumer_group_id.try_into().unwrap(),
+ )
+ .await;
+ assert!(consumer_group.is_ok());
+ }
- let topic = client
- .delete_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
- )
- .await;
- assert!(topic.is_ok());
+ if let Some(topic_id) = self.actual_topic_id {
+ let topic = client
+ .delete_topic(
+ &self.actual_stream_id.unwrap().try_into().unwrap(),
+ &topic_id.try_into().unwrap(),
+ )
+ .await;
+ assert!(topic.is_ok());
+ }
- let stream = client
- .delete_stream(&self.stream_id.try_into().unwrap())
- .await;
- assert!(stream.is_ok());
+ if let Some(stream_id) = self.actual_stream_id {
+ let stream = client
+ .delete_stream(&stream_id.try_into().unwrap())
+ .await;
+ assert!(stream.is_ok());
+ }
}
}
@@ -177,72 +187,18 @@ pub async fn should_be_successful() {
let mut iggy_cmd_test = IggyCmdTest::default();
let test_parameters = vec![
- (
- TestStreamId::Numeric,
- TestTopicId::Numeric,
- OutputFormat::Default,
- ),
- (
- TestStreamId::Named,
- TestTopicId::Numeric,
- OutputFormat::Default,
- ),
- (
- TestStreamId::Numeric,
- TestTopicId::Named,
- OutputFormat::Default,
- ),
- (
- TestStreamId::Named,
- TestTopicId::Named,
- OutputFormat::Default,
- ),
- (
- TestStreamId::Numeric,
- TestTopicId::Numeric,
- OutputFormat::List,
- ),
- (
- TestStreamId::Named,
- TestTopicId::Numeric,
- OutputFormat::List,
- ),
- (
- TestStreamId::Numeric,
- TestTopicId::Named,
- OutputFormat::List,
- ),
- (TestStreamId::Named, TestTopicId::Named, OutputFormat::List),
- (
- TestStreamId::Numeric,
- TestTopicId::Numeric,
- OutputFormat::Table,
- ),
- (
- TestStreamId::Named,
- TestTopicId::Numeric,
- OutputFormat::Table,
- ),
- (
- TestStreamId::Numeric,
- TestTopicId::Named,
- OutputFormat::Table,
- ),
- (TestStreamId::Named, TestTopicId::Named, OutputFormat::Table),
+ OutputFormat::Default,
+ OutputFormat::List,
+ OutputFormat::Table,
];
iggy_cmd_test.setup().await;
- for (using_stream_id, using_topic_id, output_format) in test_parameters {
+ for output_format in test_parameters {
iggy_cmd_test
.execute_test(TestConsumerGroupListCmd::new(
- 1,
String::from("stream"),
- 2,
String::from("topic"),
- 3,
String::from("consumer-group"),
- using_stream_id,
- using_topic_id,
output_format,
))
.await;
diff --git a/core/integration/tests/cli/consumer_offset/test_consumer_offset_get_command.rs b/core/integration/tests/cli/consumer_offset/test_consumer_offset_get_command.rs
index af0861cf..f15b765e 100644
--- a/core/integration/tests/cli/consumer_offset/test_consumer_offset_get_command.rs
+++ b/core/integration/tests/cli/consumer_offset/test_consumer_offset_get_command.rs
@@ -18,7 +18,7 @@
use crate::cli::common::{
CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestConsumerId, TestHelpCmd,
- TestStreamId, TestTopicId, USAGE_PREFIX,
+ USAGE_PREFIX,
};
use assert_cmd::assert::Assert;
use async_trait::async_trait;
@@ -30,63 +30,59 @@ use std::str::FromStr;
struct TestConsumerOffsetGetCmd {
consumer_id: u32,
consumer_name: String,
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
partition_id: u32,
using_consumer_id: TestConsumerId,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
messages_count: u32,
stored_offset: u64,
+ // These will be populated after creating the resources
+ actual_stream_id: Option<u32>,
+ actual_topic_id: Option<u32>,
}
impl TestConsumerOffsetGetCmd {
- #[allow(clippy::too_many_arguments)]
fn new(
consumer_id: u32,
consumer_name: String,
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
partition_id: u32,
using_consumer_id: TestConsumerId,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
) -> Self {
Self {
consumer_id,
consumer_name,
- stream_id,
stream_name,
- topic_id,
topic_name,
partition_id,
using_consumer_id,
- using_stream_id,
- using_topic_id,
messages_count: 100,
stored_offset: 66,
+ actual_stream_id: None,
+ actual_topic_id: None,
}
}
fn to_args(&self) -> Vec<String> {
let mut command = match self.using_consumer_id {
- TestStreamId::Numeric => vec![format!("{}", self.consumer_id)],
- TestStreamId::Named => vec![self.consumer_name.clone()],
+ TestConsumerId::Numeric => vec![format!("{}", self.consumer_id)],
+ TestConsumerId::Named => vec![self.consumer_name.clone()],
};
- command.push(match self.using_stream_id {
- TestTopicId::Numeric => format!("{}", self.stream_id),
- TestTopicId::Named => self.stream_name.clone(),
- });
+ // Use actual stream ID if available, otherwise use stream name as fallback
+ if let Some(stream_id) = self.actual_stream_id {
+ command.push(format!("{}", stream_id));
+ } else {
+ command.push(self.stream_name.clone());
+ }
- command.push(match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
- });
+ // Use actual topic ID if available, otherwise use topic name as fallback
+ if let Some(topic_id) = self.actual_topic_id {
+ command.push(format!("{}", topic_id));
+ } else {
+ command.push(self.topic_name.clone());
+ }
command.push(format!("{}", self.partition_id));
@@ -97,24 +93,27 @@ impl TestConsumerOffsetGetCmd {
#[async_trait]
impl IggyCmdTestCase for TestConsumerOffsetGetCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
+ // Create stream and capture its actual ID
let stream = client
- .create_stream(&self.stream_name, self.stream_id.into())
- .await;
- assert!(stream.is_ok());
+ .create_stream(&self.stream_name)
+ .await
+ .expect("Failed to create stream");
+ self.actual_stream_id = Some(stream.id);
+ // Create topic and capture its actual ID
let topic = client
.create_topic(
- &self.stream_id.try_into().unwrap(),
+ &stream.id.try_into().unwrap(),
&self.topic_name,
1,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
- .await;
- assert!(topic.is_ok());
+ .await
+ .expect("Failed to create topic");
+ self.actual_topic_id = Some(topic.id);
let mut messages = (1..=self.messages_count)
.filter_map(|id| IggyMessage::from_str(format!("Test message {id}").as_str()).ok())
@@ -122,8 +121,8 @@ impl IggyCmdTestCase for TestConsumerOffsetGetCmd {
let send_status = client
.send_messages(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &stream.id.try_into().unwrap(),
+ &topic.id.try_into().unwrap(),
&Partitioning::partition_id(self.partition_id),
&mut messages,
)
@@ -140,8 +139,8 @@ impl IggyCmdTestCase for TestConsumerOffsetGetCmd {
}
.unwrap(),
},
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &stream.id.try_into().unwrap(),
+ &topic.id.try_into().unwrap(),
Some(self.partition_id),
self.stored_offset,
)
@@ -163,14 +162,16 @@ impl IggyCmdTestCase for TestConsumerOffsetGetCmd {
TestConsumerId::Named => self.consumer_name.clone(),
};
- let stream_id = match self.using_stream_id {
- TestStreamId::Numeric => format!("{}", self.stream_id),
- TestStreamId::Named => self.stream_name.clone(),
+ let stream_id = if let Some(stream_id) = self.actual_stream_id {
+ format!("{}", stream_id)
+ } else {
+ self.stream_name.clone()
};
- let topic_id = match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
+ let topic_id = if let Some(topic_id) = self.actual_topic_id {
+ format!("{}", topic_id)
+ } else {
+ self.topic_name.clone()
};
let message = format!(
@@ -189,18 +190,22 @@ impl IggyCmdTestCase for TestConsumerOffsetGetCmd {
}
async fn verify_server_state(&self, client: &dyn Client) {
- let topic = client
- .delete_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
- )
- .await;
- assert!(topic.is_ok());
+ if let Some(topic_id) = self.actual_topic_id {
+ let topic = client
+ .delete_topic(
+ &self.actual_stream_id.unwrap().try_into().unwrap(),
+ &topic_id.try_into().unwrap(),
+ )
+ .await;
+ assert!(topic.is_ok());
+ }
- let stream = client
- .delete_stream(&self.stream_id.try_into().unwrap())
- .await;
- assert!(stream.is_ok());
+ if let Some(stream_id) = self.actual_stream_id {
+ let stream = client
+ .delete_stream(&stream_id.try_into().unwrap())
+ .await;
+ assert!(stream.is_ok());
+ }
}
}
@@ -209,68 +214,18 @@ impl IggyCmdTestCase for TestConsumerOffsetGetCmd {
pub async fn should_be_successful() {
let mut iggy_cmd_test = IggyCmdTest::default();
- let test_parameters = vec![
- (
- TestConsumerId::Numeric,
- TestStreamId::Numeric,
- TestTopicId::Numeric,
- ),
- (
- TestConsumerId::Named,
- TestStreamId::Numeric,
- TestTopicId::Numeric,
- ),
- (
- TestConsumerId::Numeric,
- TestStreamId::Named,
- TestTopicId::Numeric,
- ),
- (
- TestConsumerId::Numeric,
- TestStreamId::Numeric,
- TestTopicId::Named,
- ),
- (
- TestConsumerId::Named,
- TestStreamId::Named,
- TestTopicId::Numeric,
- ),
- (
- TestConsumerId::Named,
- TestStreamId::Numeric,
- TestTopicId::Named,
- ),
- (
- TestConsumerId::Numeric,
- TestStreamId::Named,
- TestTopicId::Named,
- ),
- (
- TestConsumerId::Named,
- TestStreamId::Named,
- TestTopicId::Named,
- ),
- (
- TestConsumerId::Numeric,
- TestStreamId::Numeric,
- TestTopicId::Numeric,
- ),
- ];
+ let test_parameters = vec![TestConsumerId::Numeric, TestConsumerId::Named];
iggy_cmd_test.setup().await;
- for (using_stream_id, using_topic_id, using_consumer_id) in test_parameters {
+ for using_consumer_id in test_parameters {
iggy_cmd_test
.execute_test(TestConsumerOffsetGetCmd::new(
1,
String::from("consumer"),
- 2,
String::from("stream"),
- 3,
String::from("topic"),
1,
using_consumer_id,
- using_stream_id,
- using_topic_id,
))
.await;
}
diff --git a/core/integration/tests/cli/consumer_offset/test_consumer_offset_set_command.rs b/core/integration/tests/cli/consumer_offset/test_consumer_offset_set_command.rs
index f1a107a2..4a003d22 100644
--- a/core/integration/tests/cli/consumer_offset/test_consumer_offset_set_command.rs
+++ b/core/integration/tests/cli/consumer_offset/test_consumer_offset_set_command.rs
@@ -98,7 +98,7 @@ impl TestConsumerOffsetSetCmd {
impl IggyCmdTestCase for TestConsumerOffsetSetCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client
- .create_stream(&self.stream_name, Some(self.stream_id))
+ .create_stream(&self.stream_name)
.await;
assert!(stream.is_ok());
@@ -109,7 +109,6 @@ impl IggyCmdTestCase for TestConsumerOffsetSetCmd {
1,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
diff --git a/core/integration/tests/cli/message/test_message_flush_command.rs b/core/integration/tests/cli/message/test_message_flush_command.rs
index ce0c888c..7dc295f2 100644
--- a/core/integration/tests/cli/message/test_message_flush_command.rs
+++ b/core/integration/tests/cli/message/test_message_flush_command.rs
@@ -17,8 +17,7 @@
*/
use crate::cli::common::{
- CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestHelpCmd, TestStreamId,
- TestTopicId, USAGE_PREFIX,
+ CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestHelpCmd, USAGE_PREFIX,
};
use assert_cmd::assert::Assert;
use async_trait::async_trait;
@@ -31,55 +30,51 @@ use serial_test::parallel;
use std::str::FromStr;
struct TestMessageFetchCmd {
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
partitions_count: u32,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
partition_id: u32,
fsync: bool,
+ // These will be populated after creating the resources
+ actual_stream_id: Option<u32>,
+ actual_topic_id: Option<u32>,
}
impl TestMessageFetchCmd {
- #[allow(clippy::too_many_arguments)]
fn new(
- stream_id: u32,
stream_name: &str,
- topic_id: u32,
topic_name: &str,
partitions_count: u32,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
partition_id: u32,
fsync: bool,
) -> Self {
Self {
- stream_id,
stream_name: stream_name.to_string(),
- topic_id,
topic_name: topic_name.to_string(),
partitions_count,
- using_stream_id,
- using_topic_id,
partition_id,
fsync,
+ actual_stream_id: None,
+ actual_topic_id: None,
}
}
fn to_args(&self) -> Vec<String> {
let mut command = Vec::new();
- command.extend(match self.using_stream_id {
- TestStreamId::Numeric => vec![format!("{}", self.stream_id)],
- TestStreamId::Named => vec![self.stream_name.clone()],
- });
+ // Use actual stream ID if available, otherwise use stream name as fallback
+ if let Some(stream_id) = self.actual_stream_id {
+ command.push(format!("{}", stream_id));
+ } else {
+ command.push(self.stream_name.clone());
+ }
- command.push(match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
- });
+ // Use actual topic ID if available, otherwise use topic name as fallback
+ if let Some(topic_id) = self.actual_topic_id {
+ command.push(format!("{}", topic_id));
+ } else {
+ command.push(self.topic_name.clone());
+ }
command.push(format!("{}", self.partition_id));
@@ -94,26 +89,25 @@ impl TestMessageFetchCmd {
#[async_trait]
impl IggyCmdTestCase for TestMessageFetchCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client.create_stream(&self.stream_name, None).await;
- assert!(stream.is_ok());
-
- let stream_id = Identifier::from_str(&self.stream_name);
- assert!(stream_id.is_ok());
- let stream_id = stream_id.unwrap();
+ // Create stream and capture its actual ID
+ let stream = client.create_stream(&self.stream_name).await
+ .expect("Failed to create stream");
+ self.actual_stream_id = Some(stream.id);
+ // Create topic and capture its actual ID
let topic = client
.create_topic(
- &stream_id,
+ &stream.id.try_into().unwrap(),
&self.topic_name,
self.partitions_count,
Default::default(),
None,
- None,
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
- .await;
- assert!(topic.is_ok());
+ .await
+ .expect("Failed to create topic");
+ self.actual_topic_id = Some(topic.id);
}
fn get_command(&self) -> IggyCmdCommand {
@@ -125,14 +119,16 @@ impl IggyCmdTestCase for TestMessageFetchCmd {
}
fn verify_command(&self, command_state: Assert) {
- let stream_id = match self.using_stream_id {
- TestStreamId::Numeric => format!("{}", self.stream_id),
- TestStreamId::Named => self.stream_name.clone(),
+ let stream_id = if let Some(stream_id) = self.actual_stream_id {
+ format!("{}", stream_id)
+ } else {
+ self.stream_name.clone()
};
- let topic_id = match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
+ let topic_id = if let Some(topic_id) = self.actual_topic_id {
+ format!("{}", topic_id)
+ } else {
+ self.topic_name.clone()
};
let identification_part = format!(
@@ -153,18 +149,17 @@ impl IggyCmdTestCase for TestMessageFetchCmd {
}
async fn verify_server_state(&self, client: &dyn Client) {
- let topic_id = Identifier::from_str(&self.topic_name);
- assert!(topic_id.is_ok());
- let topic_id = topic_id.unwrap();
- let stream_id = Identifier::from_str(&self.stream_name);
- assert!(stream_id.is_ok());
- let stream_id = stream_id.unwrap();
-
- let topic = client.delete_topic(&stream_id, &topic_id).await;
- assert!(topic.is_ok());
-
- let stream = client.delete_stream(&stream_id).await;
- assert!(stream.is_ok());
+ if let Some(topic_id) = self.actual_topic_id {
+ let stream_id = Identifier::from_str(&self.stream_name);
+ assert!(stream_id.is_ok());
+ let stream_id = stream_id.unwrap();
+
+ let topic = client.delete_topic(&stream_id, &topic_id.try_into().unwrap()).await;
+ assert!(topic.is_ok());
+
+ let stream = client.delete_stream(&stream_id).await;
+ assert!(stream.is_ok());
+ }
}
}
@@ -173,28 +168,15 @@ impl IggyCmdTestCase for TestMessageFetchCmd {
pub async fn should_be_successful() {
let mut iggy_cmd_test = IggyCmdTest::default();
- let test_parameters: Vec<(TestStreamId, TestTopicId, bool)> = vec![
- (TestStreamId::Numeric, TestTopicId::Numeric, false),
- (TestStreamId::Numeric, TestTopicId::Named, false),
- (TestStreamId::Named, TestTopicId::Numeric, false),
- (TestStreamId::Named, TestTopicId::Named, false),
- (TestStreamId::Numeric, TestTopicId::Numeric, true),
- (TestStreamId::Numeric, TestTopicId::Named, true),
- (TestStreamId::Named, TestTopicId::Numeric, true),
- (TestStreamId::Named, TestTopicId::Named, true),
- ];
+ let test_parameters = vec![false, true]; // Test with and without fsync
iggy_cmd_test.setup().await;
- for (using_stream_id, using_topic_id, fsync) in test_parameters {
+ for fsync in test_parameters {
iggy_cmd_test
.execute_test(TestMessageFetchCmd::new(
- 1,
"stream",
+ "topic",
1,
- "topic",
- 1,
- using_stream_id,
- using_topic_id,
1,
fsync,
))
diff --git a/core/integration/tests/cli/message/test_message_poll_command.rs b/core/integration/tests/cli/message/test_message_poll_command.rs
index aea54964..3227beff 100644
--- a/core/integration/tests/cli/message/test_message_poll_command.rs
+++ b/core/integration/tests/cli/message/test_message_poll_command.rs
@@ -17,8 +17,7 @@
*/
use crate::cli::common::{
- CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestHelpCmd, TestStreamId,
- TestTopicId, USAGE_PREFIX,
+ CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestHelpCmd, USAGE_PREFIX,
};
use assert_cmd::assert::Assert;
use async_trait::async_trait;
@@ -30,35 +29,30 @@ use std::collections::HashMap;
use std::str::FromStr;
struct TestMessagePollCmd {
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
partitions_count: u32,
messages: Vec<String>,
partition_id: u32,
message_count: usize,
strategy: PollingStrategy,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
show_headers: bool,
headers: (HeaderKey, HeaderValue),
+ // These will be populated after creating the resources
+ actual_stream_id: Option<u32>,
+ actual_topic_id: Option<u32>,
}
impl TestMessagePollCmd {
#[allow(clippy::too_many_arguments)]
fn new(
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
partitions_count: u32,
messages: &[String],
partition_id: u32,
message_count: usize,
strategy: PollingStrategy,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
show_headers: bool,
headers: (HeaderKey, HeaderValue),
) -> Self {
@@ -66,19 +60,17 @@ impl TestMessagePollCmd {
assert!(partition_id > 0);
assert!(message_count < messages.len());
Self {
- stream_id,
stream_name,
- topic_id,
topic_name,
partitions_count,
messages: messages.to_owned(),
partition_id,
message_count,
strategy,
- using_stream_id,
- using_topic_id,
show_headers,
headers,
+ actual_stream_id: None,
+ actual_topic_id: None,
}
}
@@ -102,15 +94,19 @@ impl TestMessagePollCmd {
command.extend(vec!["--show-headers".into()]);
}
- command.extend(match self.using_stream_id {
- TestStreamId::Numeric => vec![format!("{}", self.stream_id)],
- TestStreamId::Named => vec![self.stream_name.clone()],
- });
+ // Use actual stream ID if available, otherwise use stream name as fallback
+ if let Some(stream_id) = self.actual_stream_id {
+ command.push(format!("{}", stream_id));
+ } else {
+ command.push(self.stream_name.clone());
+ }
- command.push(match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
- });
+ // Use actual topic ID if available, otherwise use topic name as fallback
+ if let Some(topic_id) = self.actual_topic_id {
+ command.push(format!("{}", topic_id));
+ } else {
+ command.push(self.topic_name.clone());
+ }
command.push(format!("{}", self.partition_id));
@@ -121,25 +117,28 @@ impl TestMessagePollCmd {
#[async_trait]
impl IggyCmdTestCase for TestMessagePollCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name, self.stream_id.into())
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
+ let stream = stream.unwrap();
+ self.actual_stream_id = Some(stream.id);
+ let stream_id = Identifier::from_str(&self.stream_name).unwrap();
let topic = client
.create_topic(
- &self.stream_id.try_into().unwrap(),
+ &stream_id,
&self.topic_name,
self.partitions_count,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
.await;
assert!(topic.is_ok());
+ let topic = topic.unwrap();
+ self.actual_topic_id = Some(topic.id);
+ let topic_id = Identifier::from_str(&self.topic_name).unwrap();
let mut messages = self
.messages
.iter()
@@ -155,8 +154,8 @@ impl IggyCmdTestCase for TestMessagePollCmd {
let send_status = client
.send_messages(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &stream_id,
+ &topic_id,
&Partitioning::partition_id(self.partition_id),
&mut messages,
)
@@ -173,14 +172,16 @@ impl IggyCmdTestCase for TestMessagePollCmd {
}
fn verify_command(&self, command_state: Assert) {
- let stream_id = match self.using_stream_id {
- TestStreamId::Numeric => format!("{}", self.stream_id),
- TestStreamId::Named => self.stream_name.clone(),
+ let stream_id = if let Some(stream_id) = self.actual_stream_id {
+ format!("{}", stream_id)
+ } else {
+ self.stream_name.clone()
};
- let topic_id = match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
+ let topic_id = if let Some(topic_id) = self.actual_topic_id {
+ format!("{}", topic_id)
+ } else {
+ self.topic_name.clone()
};
let polled_status = match self.message_count {
@@ -235,17 +236,13 @@ impl IggyCmdTestCase for TestMessagePollCmd {
}
async fn verify_server_state(&self, client: &dyn Client) {
- let topic = client
- .delete_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
- )
- .await;
+ let stream_id = Identifier::from_str(&self.stream_name).unwrap();
+ let topic_id = Identifier::from_str(&self.topic_name).unwrap();
+
+ let topic = client.delete_topic(&stream_id, &topic_id).await;
assert!(topic.is_ok());
- let stream = client
- .delete_stream(&self.stream_id.try_into().unwrap())
- .await;
+ let stream = client.delete_stream(&stream_id).await;
assert!(stream.is_ok());
}
}
@@ -273,74 +270,26 @@ pub async fn should_be_successful() {
HeaderValue::from_str("value1").unwrap(),
);
- let test_parameters: Vec<(u32, usize, PollingStrategy, TestStreamId, TestTopicId, bool)> = vec![
- (
- 1,
- 1,
- PollingStrategy::offset(0),
- TestStreamId::Numeric,
- TestTopicId::Numeric,
- true,
- ),
- (
- 2,
- 5,
- PollingStrategy::offset(0),
- TestStreamId::Numeric,
- TestTopicId::Named,
- true,
- ),
- (
- 3,
- 3,
- PollingStrategy::offset(3),
- TestStreamId::Named,
- TestTopicId::Numeric,
- true,
- ),
- (
- 4,
- 5,
- PollingStrategy::first(),
- TestStreamId::Named,
- TestTopicId::Named,
- true,
- ),
- (
- 1,
- 4,
- PollingStrategy::last(),
- TestStreamId::Numeric,
- TestTopicId::Numeric,
- true,
- ),
- (
- 2,
- 3,
- PollingStrategy::next(),
- TestStreamId::Numeric,
- TestTopicId::Named,
- false,
- ),
+ let test_parameters: Vec<(u32, usize, PollingStrategy, bool)> = vec![
+ (1, 1, PollingStrategy::offset(0), true),
+ (2, 5, PollingStrategy::offset(0), true),
+ (3, 3, PollingStrategy::offset(3), true),
+ (4, 5, PollingStrategy::first(), true),
+ (1, 4, PollingStrategy::last(), true),
+ (2, 3, PollingStrategy::next(), false),
];
iggy_cmd_test.setup().await;
- for (partition_id, message_count, strategy, using_stream_id, using_topic_id, show_headers) in
- test_parameters
- {
+ for (partition_id, message_count, strategy, show_headers) in test_parameters {
iggy_cmd_test
.execute_test(TestMessagePollCmd::new(
- 1,
String::from("stream"),
- 2,
String::from("topic"),
4,
&test_messages,
partition_id,
message_count,
strategy,
- using_stream_id,
- using_topic_id,
show_headers,
test_headers.clone(),
))
diff --git a/core/integration/tests/cli/message/test_message_poll_to_file_command.rs b/core/integration/tests/cli/message/test_message_poll_to_file_command.rs
index 733871c4..ace0e485 100644
--- a/core/integration/tests/cli/message/test_message_poll_to_file_command.rs
+++ b/core/integration/tests/cli/message/test_message_poll_to_file_command.rs
@@ -94,7 +94,7 @@ impl<'a> TestMessagePollToFileCmd<'a> {
#[async_trait]
impl IggyCmdTestCase for TestMessagePollToFileCmd<'_> {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client.create_stream(&self.stream_name, None).await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
let stream_id = Identifier::from_str(self.stream_name.as_str());
@@ -108,7 +108,6 @@ impl IggyCmdTestCase for TestMessagePollToFileCmd<'_> {
1,
Default::default(),
None,
- None,
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
diff --git a/core/integration/tests/cli/message/test_message_send_command.rs b/core/integration/tests/cli/message/test_message_send_command.rs
index 2a9cf5a5..70155458 100644
--- a/core/integration/tests/cli/message/test_message_send_command.rs
+++ b/core/integration/tests/cli/message/test_message_send_command.rs
@@ -17,8 +17,7 @@
*/
use crate::cli::common::{
- CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestHelpCmd, TestStreamId,
- TestTopicId, USAGE_PREFIX,
+ CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestHelpCmd, USAGE_PREFIX,
};
use assert_cmd::assert::Assert;
use async_trait::async_trait;
@@ -53,61 +52,57 @@ enum ProvideMessages {
}
struct TestMessageSendCmd {
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
partitions_count: u32,
messages: Vec<String>,
message_input: ProvideMessages,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
partitioning: PartitionSelection,
header: Option<HashMap<HeaderKey, HeaderValue>>,
+ // These will be populated after creating the resources
+ actual_stream_id: Option<u32>,
+ actual_topic_id: Option<u32>,
}
impl TestMessageSendCmd {
- #[allow(clippy::too_many_arguments)]
fn new(
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
partitions_count: u32,
messages: Vec<String>,
message_input: ProvideMessages,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
partitioning: PartitionSelection,
header: Option<HashMap<HeaderKey, HeaderValue>>,
) -> Self {
Self {
- stream_id,
stream_name,
- topic_id,
topic_name,
partitions_count,
messages,
message_input,
- using_stream_id,
- using_topic_id,
partitioning,
header,
+ actual_stream_id: None,
+ actual_topic_id: None,
}
}
fn to_args(&self) -> Vec<String> {
let mut command = self.partitioning.to_args();
- command.extend(match self.using_stream_id {
- TestStreamId::Numeric => vec![format!("{}", self.stream_id)],
- TestStreamId::Named => vec![self.stream_name.clone()],
- });
+ // Use actual stream ID if available, otherwise use stream name as fallback
+ if let Some(stream_id) = self.actual_stream_id {
+ command.push(format!("{}", stream_id));
+ } else {
+ command.push(self.stream_name.clone());
+ }
- command.push(match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
- });
+ // Use actual topic ID if available, otherwise use topic name as fallback
+ if let Some(topic_id) = self.actual_topic_id {
+ command.push(format!("{}", topic_id));
+ } else {
+ command.push(self.topic_name.clone());
+ }
if let Some(header) = &self.header {
command.push("--headers".to_string());
@@ -154,24 +149,27 @@ impl TestMessageSendCmd {
#[async_trait]
impl IggyCmdTestCase for TestMessageSendCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
+ // Create stream and capture its actual ID
let stream = client
- .create_stream(&self.stream_name, self.stream_id.into())
- .await;
- assert!(stream.is_ok());
+ .create_stream(&self.stream_name)
+ .await
+ .expect("Failed to create stream");
+ self.actual_stream_id = Some(stream.id);
+ // Create topic and capture its actual ID
let topic = client
.create_topic(
- &self.stream_id.try_into().unwrap(),
+ &stream.id.try_into().unwrap(),
&self.topic_name,
self.partitions_count,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
- .await;
- assert!(topic.is_ok());
+ .await
+ .expect("Failed to create topic");
+ self.actual_topic_id = Some(topic.id);
}
fn get_command(&self) -> IggyCmdCommand {
@@ -190,14 +188,16 @@ impl IggyCmdTestCase for TestMessageSendCmd {
}
fn verify_command(&self, command_state: Assert) {
- let stream_id = match self.using_stream_id {
- TestStreamId::Numeric => format!("{}", self.stream_id),
- TestStreamId::Named => self.stream_name.clone(),
+ let stream_id = if let Some(stream_id) = self.actual_stream_id {
+ format!("{}", stream_id)
+ } else {
+ self.stream_name.clone()
};
- let topic_id = match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
+ let topic_id = if let Some(topic_id) = self.actual_topic_id {
+ format!("{}", topic_id)
+ } else {
+ self.topic_name.clone()
};
let message = format!(
@@ -210,8 +210,8 @@ impl IggyCmdTestCase for TestMessageSendCmd {
async fn verify_server_state(&self, client: &dyn Client) {
let topic = client
.get_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.actual_stream_id.unwrap().try_into().unwrap(),
+ &self.actual_topic_id.unwrap().try_into().unwrap(),
)
.await;
assert!(topic.is_ok());
@@ -220,8 +220,8 @@ impl IggyCmdTestCase for TestMessageSendCmd {
let polled_messages = client
.poll_messages(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.actual_stream_id.unwrap().try_into().unwrap(),
+ &self.actual_topic_id.unwrap().try_into().unwrap(),
Some(self.get_partition_id()),
&Consumer::default(),
&PollingStrategy::offset(0),
@@ -251,14 +251,14 @@ impl IggyCmdTestCase for TestMessageSendCmd {
let topic = client
.delete_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.actual_stream_id.unwrap().try_into().unwrap(),
+ &self.actual_topic_id.unwrap().try_into().unwrap(),
)
.await;
assert!(topic.is_ok());
let stream = client
- .delete_stream(&self.stream_id.try_into().unwrap())
+ .delete_stream(&self.actual_stream_id.unwrap().try_into().unwrap())
.await;
assert!(stream.is_ok());
}
@@ -270,87 +270,19 @@ pub async fn should_be_successful() {
let mut iggy_cmd_test = IggyCmdTest::default();
let test_parameters = vec![
- (
- ProvideMessages::AsArgs,
- TestStreamId::Numeric,
- TestTopicId::Numeric,
- PartitionSelection::Balanced,
- ),
- (
- ProvideMessages::ViaStdin,
- TestStreamId::Numeric,
- TestTopicId::Named,
- PartitionSelection::Balanced,
- ),
- (
- ProvideMessages::AsArgs,
- TestStreamId::Named,
- TestTopicId::Numeric,
- PartitionSelection::Balanced,
- ),
- (
- ProvideMessages::ViaStdin,
- TestStreamId::Named,
- TestTopicId::Named,
- PartitionSelection::Balanced,
- ),
- (
- ProvideMessages::ViaStdin,
- TestStreamId::Numeric,
- TestTopicId::Numeric,
- PartitionSelection::Id(1),
- ),
- (
- ProvideMessages::AsArgs,
- TestStreamId::Named,
- TestTopicId::Numeric,
- PartitionSelection::Id(2),
- ),
- (
- ProvideMessages::AsArgs,
- TestStreamId::Numeric,
- TestTopicId::Named,
- PartitionSelection::Id(3),
- ),
- (
- ProvideMessages::ViaStdin,
- TestStreamId::Named,
- TestTopicId::Named,
- PartitionSelection::Id(4),
- ),
- (
- ProvideMessages::AsArgs,
- TestStreamId::Numeric,
- TestTopicId::Numeric,
- PartitionSelection::Key(String::from("some-complex-key")),
- ),
- (
- ProvideMessages::ViaStdin,
- TestStreamId::Named,
- TestTopicId::Numeric,
- PartitionSelection::Key(String::from("another random value")),
- ),
- (
- ProvideMessages::ViaStdin,
- TestStreamId::Numeric,
- TestTopicId::Named,
- PartitionSelection::Key(String::from("some-random-key")),
- ),
- (
- ProvideMessages::AsArgs,
- TestStreamId::Named,
- TestTopicId::Named,
- PartitionSelection::Key(String::from("just a key")),
- ),
+ (ProvideMessages::AsArgs, PartitionSelection::Balanced),
+ (ProvideMessages::ViaStdin, PartitionSelection::Balanced),
+ (ProvideMessages::ViaStdin, PartitionSelection::Id(1)),
+ (ProvideMessages::AsArgs, PartitionSelection::Id(2)),
+ (ProvideMessages::AsArgs, PartitionSelection::Key(String::from("some-complex-key"))),
+ (ProvideMessages::ViaStdin, PartitionSelection::Key(String::from("another-key"))),
];
iggy_cmd_test.setup().await;
- for (message_input, using_stream_id, using_topic_id, using_partitioning) in test_parameters {
+ for (message_input, using_partitioning) in test_parameters {
iggy_cmd_test
.execute_test(TestMessageSendCmd::new(
- 1,
String::from("stream"),
- 2,
String::from("topic"),
4,
vec![
@@ -359,8 +291,6 @@ pub async fn should_be_successful() {
String::from("test message3"),
],
message_input,
- using_stream_id,
- using_topic_id,
using_partitioning,
Some(HashMap::from([
(
diff --git a/core/integration/tests/cli/message/test_message_send_from_file_command.rs b/core/integration/tests/cli/message/test_message_send_from_file_command.rs
index 19cf4c55..3327a686 100644
--- a/core/integration/tests/cli/message/test_message_send_from_file_command.rs
+++ b/core/integration/tests/cli/message/test_message_send_from_file_command.rs
@@ -77,7 +77,7 @@ impl<'a> TestMessageSendFromFileCmd<'a> {
#[async_trait]
impl IggyCmdTestCase for TestMessageSendFromFileCmd<'_> {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client.create_stream(&self.stream_name, None).await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
let stream_id = Identifier::from_str(self.stream_name.as_str());
@@ -91,7 +91,6 @@ impl IggyCmdTestCase for TestMessageSendFromFileCmd<'_> {
1,
Default::default(),
None,
- None,
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
diff --git a/core/integration/tests/cli/partition/test_partition_create_command.rs b/core/integration/tests/cli/partition/test_partition_create_command.rs
index e1f0b191..d1ced0b7 100644
--- a/core/integration/tests/cli/partition/test_partition_create_command.rs
+++ b/core/integration/tests/cli/partition/test_partition_create_command.rs
@@ -17,8 +17,7 @@
*/
use crate::cli::common::{
- CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestHelpCmd, TestStreamId,
- TestTopicId, USAGE_PREFIX,
+ CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestHelpCmd, USAGE_PREFIX,
};
use assert_cmd::assert::Assert;
use async_trait::async_trait;
@@ -30,53 +29,51 @@ use predicates::str::diff;
use serial_test::parallel;
struct TestPartitionCreateCmd {
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
partitions_count: u32,
compression_algorithm: CompressionAlgorithm,
new_partitions: u32,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
+ // These will be populated after creating the resources
+ actual_stream_id: Option<u32>,
+ actual_topic_id: Option<u32>,
}
impl TestPartitionCreateCmd {
- #[allow(clippy::too_many_arguments)]
fn new(
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
partitions_count: u32,
compression_algorithm: CompressionAlgorithm,
new_partitions: u32,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
) -> Self {
Self {
- stream_id,
stream_name,
- topic_id,
topic_name,
partitions_count,
compression_algorithm,
new_partitions,
- using_stream_id,
- using_topic_id,
+ actual_stream_id: None,
+ actual_topic_id: None,
}
}
fn to_args(&self) -> Vec<String> {
- let mut command = match self.using_stream_id {
- TestStreamId::Numeric => vec![format!("{}", self.stream_id)],
- TestStreamId::Named => vec![self.stream_name.clone()],
- };
+ let mut command = vec![];
+
+ // Use actual stream ID if available, otherwise use stream name as fallback
+ if let Some(stream_id) = self.actual_stream_id {
+ command.push(format!("{}", stream_id));
+ } else {
+ command.push(self.stream_name.clone());
+ }
- command.push(match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
- });
+ // Use actual topic ID if available, otherwise use topic name as fallback
+ if let Some(topic_id) = self.actual_topic_id {
+ command.push(format!("{}", topic_id));
+ } else {
+ command.push(self.topic_name.clone());
+ }
command.push(format!("{}", self.new_partitions));
@@ -87,24 +84,23 @@ impl TestPartitionCreateCmd {
#[async_trait]
impl IggyCmdTestCase for TestPartitionCreateCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name, self.stream_id.into())
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
+ self.actual_stream_id = Some(stream.unwrap().id);
let topic = client
.create_topic(
- &self.stream_id.try_into().unwrap(),
+ &self.actual_stream_id.unwrap().try_into().unwrap(),
&self.topic_name,
self.partitions_count,
self.compression_algorithm,
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
.await;
assert!(topic.is_ok());
+ self.actual_topic_id = Some(topic.unwrap().id);
}
fn get_command(&self) -> IggyCmdCommand {
@@ -116,14 +112,16 @@ impl IggyCmdTestCase for TestPartitionCreateCmd {
}
fn verify_command(&self, command_state: Assert) {
- let stream_id = match self.using_stream_id {
- TestStreamId::Numeric => format!("{}", self.stream_id),
- TestStreamId::Named => self.stream_name.clone(),
+ let stream_id = if let Some(stream_id) = self.actual_stream_id {
+ format!("{}", stream_id)
+ } else {
+ self.stream_name.clone()
};
- let topic_id = match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
+ let topic_id = if let Some(topic_id) = self.actual_topic_id {
+ format!("{}", topic_id)
+ } else {
+ self.topic_name.clone()
};
let mut partitions = String::from("partition");
@@ -142,14 +140,14 @@ impl IggyCmdTestCase for TestPartitionCreateCmd {
async fn verify_server_state(&self, client: &dyn Client) {
let topic = client
.get_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.actual_stream_id.unwrap().try_into().unwrap(),
+ &self.actual_topic_id.unwrap().try_into().unwrap(),
)
.await;
assert!(topic.is_ok());
let topic_details = topic.unwrap().expect("Failed to get topic");
assert_eq!(topic_details.name, self.topic_name);
- assert_eq!(topic_details.id, self.topic_id);
+ assert_eq!(topic_details.id, self.actual_topic_id.unwrap());
assert_eq!(
topic_details.partitions_count,
self.partitions_count + self.new_partitions
@@ -158,14 +156,14 @@ impl IggyCmdTestCase for TestPartitionCreateCmd {
let topic = client
.delete_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.actual_stream_id.unwrap().try_into().unwrap(),
+ &self.actual_topic_id.unwrap().try_into().unwrap(),
)
.await;
assert!(topic.is_ok());
let stream = client
- .delete_stream(&self.stream_id.try_into().unwrap())
+ .delete_stream(&self.actual_stream_id.unwrap().try_into().unwrap())
.await;
assert!(stream.is_ok());
}
@@ -179,54 +177,38 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestPartitionCreateCmd::new(
- 1,
String::from("main"),
- 1,
String::from("sync"),
1,
Default::default(),
1,
- TestStreamId::Numeric,
- TestTopicId::Numeric,
))
.await;
iggy_cmd_test
.execute_test(TestPartitionCreateCmd::new(
- 2,
String::from("stream"),
- 3,
String::from("topic"),
3,
Default::default(),
2,
- TestStreamId::Named,
- TestTopicId::Numeric,
))
.await;
iggy_cmd_test
.execute_test(TestPartitionCreateCmd::new(
- 4,
String::from("development"),
- 1,
String::from("probe"),
0,
Default::default(),
4,
- TestStreamId::Numeric,
- TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestPartitionCreateCmd::new(
- 2,
String::from("production"),
- 5,
String::from("test"),
4,
Default::default(),
1,
- TestStreamId::Named,
- TestTopicId::Named,
))
.await;
}
diff --git a/core/integration/tests/cli/partition/test_partition_delete_command.rs b/core/integration/tests/cli/partition/test_partition_delete_command.rs
index 0b789aee..1fdfc3c3 100644
--- a/core/integration/tests/cli/partition/test_partition_delete_command.rs
+++ b/core/integration/tests/cli/partition/test_partition_delete_command.rs
@@ -17,8 +17,7 @@
*/
use crate::cli::common::{
- CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestHelpCmd, TestStreamId,
- TestTopicId, USAGE_PREFIX,
+ CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestHelpCmd, USAGE_PREFIX,
};
use assert_cmd::assert::Assert;
use async_trait::async_trait;
@@ -29,50 +28,48 @@ use predicates::str::diff;
use serial_test::parallel;
struct TestPartitionDeleteCmd {
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
partitions_count: u32,
new_partitions: u32,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
+ // These will be populated after creating the resources
+ actual_stream_id: Option<u32>,
+ actual_topic_id: Option<u32>,
}
impl TestPartitionDeleteCmd {
- #[allow(clippy::too_many_arguments)]
fn new(
- stream_id: u32,
stream_name: String,
- topic_id: u32,
topic_name: String,
partitions_count: u32,
new_partitions: u32,
- using_stream_id: TestStreamId,
- using_topic_id: TestTopicId,
) -> Self {
Self {
- stream_id,
stream_name,
- topic_id,
topic_name,
partitions_count,
new_partitions,
- using_stream_id,
- using_topic_id,
+ actual_stream_id: None,
+ actual_topic_id: None,
}
}
fn to_args(&self) -> Vec<String> {
- let mut command = match self.using_stream_id {
- TestStreamId::Numeric => vec![format!("{}", self.stream_id)],
- TestStreamId::Named => vec![self.stream_name.clone()],
- };
+ let mut command = vec![];
+
+ // Use actual stream ID if available, otherwise use stream name as fallback
+ if let Some(stream_id) = self.actual_stream_id {
+ command.push(format!("{}", stream_id));
+ } else {
+ command.push(self.stream_name.clone());
+ }
- command.push(match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
- });
+ // Use actual topic ID if available, otherwise use topic name as fallback
+ if let Some(topic_id) = self.actual_topic_id {
+ command.push(format!("{}", topic_id));
+ } else {
+ command.push(self.topic_name.clone());
+ }
command.push(format!("{}", self.new_partitions));
@@ -83,24 +80,23 @@ impl TestPartitionDeleteCmd {
#[async_trait]
impl IggyCmdTestCase for TestPartitionDeleteCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client
- .create_stream(&self.stream_name, self.stream_id.into())
- .await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
+ self.actual_stream_id = Some(stream.unwrap().id);
let topic = client
.create_topic(
- &self.stream_id.try_into().unwrap(),
+ &self.actual_stream_id.unwrap().try_into().unwrap(),
&self.topic_name,
self.partitions_count,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
.await;
assert!(topic.is_ok());
+ self.actual_topic_id = Some(topic.unwrap().id);
}
fn get_command(&self) -> IggyCmdCommand {
@@ -112,14 +108,16 @@ impl IggyCmdTestCase for TestPartitionDeleteCmd {
}
fn verify_command(&self, command_state: Assert) {
- let stream_id = match self.using_stream_id {
- TestStreamId::Numeric => format!("{}", self.stream_id),
- TestStreamId::Named => self.stream_name.clone(),
+ let stream_id = if let Some(stream_id) = self.actual_stream_id {
+ format!("{}", stream_id)
+ } else {
+ self.stream_name.clone()
};
- let topic_id = match self.using_topic_id {
- TestTopicId::Numeric => format!("{}", self.topic_id),
- TestTopicId::Named => self.topic_name.clone(),
+ let topic_id = if let Some(topic_id) = self.actual_topic_id {
+ format!("{}", topic_id)
+ } else {
+ self.topic_name.clone()
};
let mut partitions = String::from("partition");
@@ -138,14 +136,14 @@ impl IggyCmdTestCase for TestPartitionDeleteCmd {
async fn verify_server_state(&self, client: &dyn Client) {
let topic = client
.get_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.actual_stream_id.unwrap().try_into().unwrap(),
+ &self.actual_topic_id.unwrap().try_into().unwrap(),
)
.await;
assert!(topic.is_ok());
let topic_details = topic.unwrap().expect("Failed to get topic");
assert_eq!(topic_details.name, self.topic_name);
- assert_eq!(topic_details.id, self.topic_id);
+ assert_eq!(topic_details.id, self.actual_topic_id.unwrap());
if self.new_partitions > self.partitions_count {
assert_eq!(topic_details.partitions_count, 0);
} else {
@@ -158,14 +156,14 @@ impl IggyCmdTestCase for TestPartitionDeleteCmd {
let topic = client
.delete_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.actual_stream_id.unwrap().try_into().unwrap(),
+ &self.actual_topic_id.unwrap().try_into().unwrap(),
)
.await;
assert!(topic.is_ok());
let stream = client
- .delete_stream(&self.stream_id.try_into().unwrap())
+ .delete_stream(&self.actual_stream_id.unwrap().try_into().unwrap())
.await;
assert!(stream.is_ok());
}
@@ -179,50 +177,34 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestPartitionDeleteCmd::new(
- 1,
String::from("main"),
- 1,
String::from("sync"),
3,
1,
- TestStreamId::Numeric,
- TestTopicId::Numeric,
))
.await;
iggy_cmd_test
.execute_test(TestPartitionDeleteCmd::new(
- 2,
String::from("stream"),
- 3,
String::from("topic"),
3,
2,
- TestStreamId::Named,
- TestTopicId::Numeric,
))
.await;
iggy_cmd_test
.execute_test(TestPartitionDeleteCmd::new(
- 4,
String::from("development"),
- 1,
String::from("probe"),
3,
3,
- TestStreamId::Numeric,
- TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestPartitionDeleteCmd::new(
- 2,
String::from("production"),
- 5,
String::from("test"),
3,
7,
- TestStreamId::Named,
- TestTopicId::Named,
))
.await;
}
diff --git a/core/integration/tests/cli/stream/test_stream_delete_command.rs b/core/integration/tests/cli/stream/test_stream_delete_command.rs
index fa0fe6a6..c1f3ebc4 100644
--- a/core/integration/tests/cli/stream/test_stream_delete_command.rs
+++ b/core/integration/tests/cli/stream/test_stream_delete_command.rs
@@ -52,7 +52,7 @@ impl TestStreamDeleteCmd {
#[async_trait]
impl IggyCmdTestCase for TestStreamDeleteCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client.create_stream(&self.name, Some(self.stream_id)).await;
+ let stream = client.create_stream(&self.name).await;
assert!(stream.is_ok());
}
diff --git a/core/integration/tests/cli/stream/test_stream_get_command.rs b/core/integration/tests/cli/stream/test_stream_get_command.rs
index 9b8c9511..ff9f5cc4 100644
--- a/core/integration/tests/cli/stream/test_stream_get_command.rs
+++ b/core/integration/tests/cli/stream/test_stream_get_command.rs
@@ -52,7 +52,7 @@ impl TestStreamGetCmd {
#[async_trait]
impl IggyCmdTestCase for TestStreamGetCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client.create_stream(&self.name, Some(self.stream_id)).await;
+ let stream = client.create_stream(&self.name).await;
assert!(stream.is_ok());
}
diff --git a/core/integration/tests/cli/stream/test_stream_list_command.rs b/core/integration/tests/cli/stream/test_stream_list_command.rs
index 3c4d4db5..5f879fc0 100644
--- a/core/integration/tests/cli/stream/test_stream_list_command.rs
+++ b/core/integration/tests/cli/stream/test_stream_list_command.rs
@@ -49,7 +49,7 @@ impl TestStreamListCmd {
#[async_trait]
impl IggyCmdTestCase for TestStreamListCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client.create_stream(&self.name, Some(self.stream_id)).await;
+ let stream = client.create_stream(&self.name).await;
assert!(stream.is_ok());
}
diff --git a/core/integration/tests/cli/stream/test_stream_purge_command.rs b/core/integration/tests/cli/stream/test_stream_purge_command.rs
index c24edff9..10358da7 100644
--- a/core/integration/tests/cli/stream/test_stream_purge_command.rs
+++ b/core/integration/tests/cli/stream/test_stream_purge_command.rs
@@ -58,7 +58,7 @@ impl TestStreamPurgeCmd {
impl IggyCmdTestCase for TestStreamPurgeCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client
- .create_stream(&self.stream_name, Some(self.stream_id))
+ .create_stream(&self.stream_name)
.await;
assert!(stream.is_ok());
@@ -69,7 +69,6 @@ impl IggyCmdTestCase for TestStreamPurgeCmd {
10,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
diff --git a/core/integration/tests/cli/stream/test_stream_update_command.rs b/core/integration/tests/cli/stream/test_stream_update_command.rs
index d7fd95fe..bf5f7d93 100644
--- a/core/integration/tests/cli/stream/test_stream_update_command.rs
+++ b/core/integration/tests/cli/stream/test_stream_update_command.rs
@@ -56,7 +56,7 @@ impl TestStreamUpdateCmd {
#[async_trait]
impl IggyCmdTestCase for TestStreamUpdateCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
- let stream = client.create_stream(&self.name, Some(self.stream_id)).await;
+ let stream = client.create_stream(&self.name).await;
assert!(stream.is_ok());
}
diff --git a/core/integration/tests/cli/system/test_stats_command.rs b/core/integration/tests/cli/system/test_stats_command.rs
index 6af7e43c..bc9b6db2 100644
--- a/core/integration/tests/cli/system/test_stats_command.rs
+++ b/core/integration/tests/cli/system/test_stats_command.rs
@@ -64,7 +64,7 @@ impl TestStatsCmd {
impl IggyCmdTestCase for TestStatsCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream_id = Identifier::from_str_value("logs").unwrap();
- let stream = client.create_stream(&stream_id.as_string(), Some(1)).await;
+ let stream = client.create_stream(&stream_id.as_string()).await;
assert!(stream.is_ok());
let topic = client
@@ -74,7 +74,6 @@ impl IggyCmdTestCase for TestStatsCmd {
5,
Default::default(),
None,
- Some(1),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
diff --git a/core/integration/tests/cli/topic/test_topic_create_command.rs b/core/integration/tests/cli/topic/test_topic_create_command.rs
index f4dc9ad5..e485bc68 100644
--- a/core/integration/tests/cli/topic/test_topic_create_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_create_command.rs
@@ -102,7 +102,7 @@ impl TestTopicCreateCmd {
impl IggyCmdTestCase for TestTopicCreateCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client
- .create_stream(&self.stream_name, Some(self.stream_id))
+ .create_stream(&self.stream_name)
.await;
assert!(stream.is_ok());
}
diff --git a/core/integration/tests/cli/topic/test_topic_delete_command.rs b/core/integration/tests/cli/topic/test_topic_delete_command.rs
index 09ed03f0..3c4f0edc 100644
--- a/core/integration/tests/cli/topic/test_topic_delete_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_delete_command.rs
@@ -75,7 +75,7 @@ impl TestTopicDeleteCmd {
impl IggyCmdTestCase for TestTopicDeleteCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client
- .create_stream(&self.stream_name, Some(self.stream_id))
+ .create_stream(&self.stream_name)
.await;
assert!(stream.is_ok());
@@ -86,7 +86,6 @@ impl IggyCmdTestCase for TestTopicDeleteCmd {
1,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
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 0a40a3c1..45b0fd01 100644
--- a/core/integration/tests/cli/topic/test_topic_get_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_get_command.rs
@@ -75,7 +75,7 @@ impl TestTopicGetCmd {
impl IggyCmdTestCase for TestTopicGetCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client
- .create_stream(&self.stream_name, Some(self.stream_id))
+ .create_stream(&self.stream_name)
.await;
assert!(stream.is_ok());
@@ -86,7 +86,6 @@ impl IggyCmdTestCase for TestTopicGetCmd {
1,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
diff --git a/core/integration/tests/cli/topic/test_topic_list_command.rs b/core/integration/tests/cli/topic/test_topic_list_command.rs
index 51669517..0b91ca58 100644
--- a/core/integration/tests/cli/topic/test_topic_list_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_list_command.rs
@@ -72,7 +72,7 @@ impl TestTopicListCmd {
impl IggyCmdTestCase for TestTopicListCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client
- .create_stream(&self.stream_name, Some(self.stream_id))
+ .create_stream(&self.stream_name)
.await;
assert!(stream.is_ok());
@@ -83,7 +83,6 @@ impl IggyCmdTestCase for TestTopicListCmd {
1,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
diff --git a/core/integration/tests/cli/topic/test_topic_purge_command.rs b/core/integration/tests/cli/topic/test_topic_purge_command.rs
index b8833989..ae310a7f 100644
--- a/core/integration/tests/cli/topic/test_topic_purge_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_purge_command.rs
@@ -74,7 +74,7 @@ impl TestTopicPurgeCmd {
impl IggyCmdTestCase for TestTopicPurgeCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client
- .create_stream(&self.stream_name, Some(self.stream_id))
+ .create_stream(&self.stream_name)
.await;
assert!(stream.is_ok());
@@ -85,7 +85,6 @@ impl IggyCmdTestCase for TestTopicPurgeCmd {
10,
Default::default(),
None,
- Some(self.topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
diff --git a/core/integration/tests/cli/topic/test_topic_update_command.rs b/core/integration/tests/cli/topic/test_topic_update_command.rs
index ac402345..9ed19bd5 100644
--- a/core/integration/tests/cli/topic/test_topic_update_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_update_command.rs
@@ -123,7 +123,7 @@ impl TestTopicUpdateCmd {
impl IggyCmdTestCase for TestTopicUpdateCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client
- .create_stream(&self.stream_name, Some(self.stream_id))
+ .create_stream(&self.stream_name)
.await;
assert!(stream.is_ok());
@@ -144,7 +144,6 @@ impl IggyCmdTestCase for TestTopicUpdateCmd {
1,
self.compression_algorithm,
Some(self.replication_factor),
- Some(self.topic_id),
message_expiry,
self.max_topic_size,
)
diff --git a/core/integration/tests/mcp/mod.rs b/core/integration/tests/mcp/mod.rs
index 1a7cd1ef..635b8899 100644
--- a/core/integration/tests/mcp/mod.rs
+++ b/core/integration/tests/mcp/mod.rs
@@ -602,7 +602,7 @@ async fn seed_data(iggy_server_address: &str) {
.expect("Failed to initialize Iggy client");
iggy_client
- .create_stream(STREAM_NAME, None)
+ .create_stream(STREAM_NAME)
.await
.expect("Failed to create stream");
@@ -613,7 +613,6 @@ async fn seed_data(iggy_server_address: &str) {
1,
iggy_common::CompressionAlgorithm::None,
None,
- None,
IggyExpiry::ServerDefault,
MaxTopicSize::ServerDefault,
)
@@ -646,7 +645,7 @@ async fn seed_data(iggy_server_address: &str) {
.expect("Failed to store consumer offset");
iggy_client
- .create_consumer_group(&STREAM_ID, &TOPIC_ID, CONSUMER_GROUP_NAME, None)
+ .create_consumer_group(&STREAM_ID, &TOPIC_ID, CONSUMER_GROUP_NAME)
.await
.expect("Failed to create consumer group");
diff --git a/core/integration/tests/sdk/producer/background.rs b/core/integration/tests/sdk/producer/background.rs
index 17af1cf2..54745e18 100644
--- a/core/integration/tests/sdk/producer/background.rs
+++ b/core/integration/tests/sdk/producer/background.rs
@@ -17,7 +17,7 @@
*/
use crate::sdk::producer::{
- PARTITION_ID, STREAM_ID, TOPIC_ID, cleanup, create_message_payload, init_system,
+ PARTITION_ID, STREAM_NAME, TOPIC_NAME, cleanup, create_message_payload, init_system,
};
use bytes::Bytes;
use iggy::clients::producer_config::BackpressureMode;
@@ -68,7 +68,7 @@ async fn background_send_receive_ok() {
}
let producer = client
- .producer(&STREAM_ID.to_string(), &TOPIC_ID.to_string())
+ .producer(STREAM_NAME, TOPIC_NAME)
.unwrap()
.partitioning(Partitioning::partition_id(PARTITION_ID))
.background(BackgroundConfig::builder().build())
@@ -81,8 +81,8 @@ async fn background_send_receive_ok() {
let consumer = Consumer::default();
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
&consumer,
&PollingStrategy::offset(0),
@@ -123,7 +123,7 @@ async fn background_buffer_overflow_immediate() {
.failure_mode(BackpressureMode::FailImmediately)
.build();
let producer = client
- .producer(&STREAM_ID.to_string(), &TOPIC_ID.to_string())
+ .producer(STREAM_NAME, TOPIC_NAME)
.unwrap()
.background(cfg)
.build();
@@ -185,7 +185,7 @@ async fn background_block_with_timeout() {
)))
.build();
let producer = client
- .producer(&STREAM_ID.to_string(), &TOPIC_ID.to_string())
+ .producer(STREAM_NAME, TOPIC_NAME)
.unwrap()
.background(cfg)
.build();
@@ -238,7 +238,7 @@ async fn background_block_waits_then_succeeds() {
.build();
let producer = client
- .producer(&STREAM_ID.to_string(), &TOPIC_ID.to_string())
+ .producer(STREAM_NAME, TOPIC_NAME)
.unwrap()
.background(cfg)
.build();
@@ -290,7 +290,7 @@ async fn background_graceful_shutdown() {
.linger_time(IggyDuration::from(2_000_000)) // 2s – long enough not to flush automatically
.build();
let producer = client
- .producer(&STREAM_ID.to_string(), &TOPIC_ID.to_string())
+ .producer(STREAM_NAME, TOPIC_NAME)
.unwrap()
.background(cfg)
.build();
@@ -307,8 +307,8 @@ async fn background_graceful_shutdown() {
let consumer = Consumer::default();
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
&consumer,
&PollingStrategy::offset(0),
@@ -322,8 +322,8 @@ async fn background_graceful_shutdown() {
producer.shutdown().await;
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
&consumer,
&PollingStrategy::offset(0),
@@ -363,7 +363,7 @@ async fn background_many_parallel_producers() {
let handle = tokio::spawn(async move {
let cfg = BackgroundConfig::builder().num_shards(1).build();
let producer = client_clone
- .producer(&STREAM_ID.to_string(), &TOPIC_ID.to_string())
+ .producer(STREAM_NAME, TOPIC_NAME)
.unwrap()
.partitioning(Partitioning::partition_id(PARTITION_ID))
.background(cfg)
@@ -391,8 +391,8 @@ async fn background_many_parallel_producers() {
let consumer = Consumer::default();
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
&consumer,
&PollingStrategy::offset(0),
diff --git a/core/integration/tests/sdk/producer/mod.rs b/core/integration/tests/sdk/producer/mod.rs
index ae7d48cc..464fdffa 100644
--- a/core/integration/tests/sdk/producer/mod.rs
+++ b/core/integration/tests/sdk/producer/mod.rs
@@ -22,8 +22,6 @@ use bytes::Bytes;
use iggy::clients::client::IggyClient;
use iggy::prelude::*;
-const STREAM_ID: u32 = 1;
-const TOPIC_ID: u32 = 1;
const PARTITION_ID: u32 = 1;
const STREAM_NAME: &str = "test-stream-producer";
const TOPIC_NAME: &str = "test-topic-producer";
@@ -36,19 +34,18 @@ fn create_message_payload(offset: u64) -> Bytes {
async fn init_system(client: &IggyClient) {
// 1. Create the stream
client
- .create_stream(STREAM_NAME, Some(STREAM_ID))
+ .create_stream(STREAM_NAME)
.await
.unwrap();
// 2. Create the topic
client
.create_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
TOPIC_NAME,
PARTITIONS_COUNT,
CompressionAlgorithm::default(),
None,
- Some(TOPIC_ID),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -58,7 +55,7 @@ async fn init_system(client: &IggyClient) {
async fn cleanup(system_client: &IggyClient) {
system_client
- .delete_stream(&Identifier::numeric(STREAM_ID).unwrap())
+ .delete_stream(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap();
}
diff --git a/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs
index 9a79bb3c..1b297228 100644
--- a/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs
@@ -17,7 +17,7 @@
*/
use crate::server::scenarios::{
- CONSUMER_GROUP_ID, CONSUMER_GROUP_NAME, PARTITIONS_COUNT, STREAM_ID, STREAM_NAME, TOPIC_ID,
+ CONSUMER_GROUP_NAME, PARTITIONS_COUNT, STREAM_NAME,
TOPIC_NAME, USERNAME_1, USERNAME_2, USERNAME_3, cleanup, create_client, join_consumer_group,
};
use iggy::clients::client::IggyClient;
@@ -50,7 +50,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 2. Create the topic
system_client
.create_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
TOPIC_NAME,
PARTITIONS_COUNT,
CompressionAlgorithm::default(),
@@ -65,8 +65,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 3. Create the consumer group
system_client
.create_consumer_group(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
CONSUMER_GROUP_NAME,
Some(CONSUMER_GROUP_ID),
)
@@ -156,9 +156,9 @@ async fn get_consumer_group_and_validate_members(
) -> ConsumerGroupDetails {
let consumer_group = client
.get_consumer_group(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
- &Identifier::numeric(CONSUMER_GROUP_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
)
.await
.unwrap()
diff --git a/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs
index ffc96a24..1d3bc543 100644
--- a/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs
@@ -17,8 +17,8 @@
*/
use crate::server::scenarios::{
- CONSUMER_GROUP_ID, CONSUMER_GROUP_NAME, MESSAGES_COUNT, PARTITIONS_COUNT, STREAM_ID,
- STREAM_NAME, TOPIC_ID, TOPIC_NAME, cleanup, create_client, get_consumer_group,
+ CONSUMER_GROUP_NAME, MESSAGES_COUNT, PARTITIONS_COUNT,
+ STREAM_NAME, TOPIC_NAME, cleanup, create_client, get_consumer_group,
join_consumer_group,
};
use iggy::prelude::*;
@@ -58,7 +58,7 @@ async fn init_system(
// 2. Create the topic
system_client
.create_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
TOPIC_NAME,
PARTITIONS_COUNT,
CompressionAlgorithm::default(),
@@ -73,8 +73,8 @@ async fn init_system(
// 3. Create the consumer group
system_client
.create_consumer_group(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
CONSUMER_GROUP_NAME,
Some(CONSUMER_GROUP_ID),
)
@@ -117,8 +117,8 @@ async fn execute_using_messages_key_key(
let mut messages = vec![message];
system_client
.send_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&Partitioning::messages_key_u32(entity_id),
&mut messages,
)
@@ -141,8 +141,8 @@ async fn poll_messages(client: &IggyClient) -> u32 {
for _ in 1..=PARTITIONS_COUNT * MESSAGES_COUNT {
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
None,
&consumer,
&PollingStrategy::next(),
@@ -181,8 +181,8 @@ async fn execute_using_none_key(
let mut messages = vec![message];
system_client
.send_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&Partitioning::balanced(),
&mut messages,
)
@@ -218,8 +218,8 @@ async fn validate_message_polling(client: &IggyClient, consumer_group: &Consumer
for i in 1..=MESSAGES_COUNT {
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
None,
&consumer,
&PollingStrategy::next(),
@@ -242,8 +242,8 @@ async fn validate_message_polling(client: &IggyClient, consumer_group: &Consumer
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
None,
&consumer,
&PollingStrategy::next(),
diff --git a/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
index 319e3edb..4933a7dc 100644
--- a/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
@@ -17,8 +17,8 @@
*/
use crate::server::scenarios::{
- CONSUMER_GROUP_ID, CONSUMER_GROUP_NAME, MESSAGES_COUNT, PARTITIONS_COUNT, STREAM_ID,
- STREAM_NAME, TOPIC_ID, TOPIC_NAME, cleanup, create_client, get_consumer_group,
+ CONSUMER_GROUP_NAME, MESSAGES_COUNT, PARTITIONS_COUNT,
+ STREAM_NAME, TOPIC_NAME, cleanup, create_client, get_consumer_group,
join_consumer_group,
};
use iggy::prelude::*;
@@ -31,7 +31,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
init_system(&client).await;
execute_using_messages_key_key(&client).await;
client
- .delete_stream(&Identifier::numeric(STREAM_ID).unwrap())
+ .delete_stream(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap();
init_system(&client).await;
@@ -43,19 +43,18 @@ pub async fn run(client_factory: &dyn ClientFactory) {
async fn init_system(client: &IggyClient) {
// 1. Create the stream
client
- .create_stream(STREAM_NAME, Some(STREAM_ID))
+ .create_stream(STREAM_NAME)
.await
.unwrap();
// 2. Create the topic
client
.create_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
TOPIC_NAME,
PARTITIONS_COUNT,
CompressionAlgorithm::default(),
None,
- Some(TOPIC_ID),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -65,10 +64,9 @@ async fn init_system(client: &IggyClient) {
// 3. Create the consumer group
client
.create_consumer_group(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
CONSUMER_GROUP_NAME,
- Some(CONSUMER_GROUP_ID),
)
.await
.unwrap();
@@ -97,8 +95,8 @@ async fn execute_using_messages_key_key(client: &IggyClient) {
let mut messages = vec![message];
client
.send_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&Partitioning::messages_key_u32(entity_id),
&mut messages,
)
@@ -107,13 +105,13 @@ async fn execute_using_messages_key_key(client: &IggyClient) {
}
// 2. Poll the messages for the single client which has assigned all partitions in the consumer group
- let consumer = Consumer::group(Identifier::numeric(CONSUMER_GROUP_ID).unwrap());
+ let consumer = Consumer::group(Identifier::named(CONSUMER_GROUP_NAME).unwrap());
let mut total_read_messages_count = 0;
for _ in 1..=PARTITIONS_COUNT * MESSAGES_COUNT {
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
None,
&consumer,
&PollingStrategy::next(),
@@ -147,8 +145,8 @@ async fn execute_using_none_key(client: &IggyClient) {
let mut messages = vec![message];
client
.send_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&Partitioning::balanced(),
&mut messages,
)
@@ -157,15 +155,15 @@ async fn execute_using_none_key(client: &IggyClient) {
}
// 2. Poll the messages for the single client which has assigned all partitions in the consumer group
- let consumer = Consumer::group(Identifier::numeric(CONSUMER_GROUP_ID).unwrap());
+ let consumer = Consumer::group(Identifier::named(CONSUMER_GROUP_NAME).unwrap());
let mut partition_id = 1;
let mut offset = 0;
let mut entity_id = 1;
for i in 1..=PARTITIONS_COUNT * MESSAGES_COUNT {
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
None,
&consumer,
&PollingStrategy::next(),
@@ -198,8 +196,8 @@ async fn execute_using_none_key(client: &IggyClient) {
for _ in 1..=PARTITIONS_COUNT {
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
None,
&consumer,
&PollingStrategy::next(),
diff --git a/core/integration/tests/server/scenarios/create_message_payload.rs b/core/integration/tests/server/scenarios/create_message_payload.rs
index e452a73f..00f84e82 100644
--- a/core/integration/tests/server/scenarios/create_message_payload.rs
+++ b/core/integration/tests/server/scenarios/create_message_payload.rs
@@ -22,8 +22,6 @@ use integration::test_server::{ClientFactory, assert_clean_system, login_root};
use std::collections::HashMap;
use std::str::FromStr;
-const STREAM_ID: u32 = 1;
-const TOPIC_ID: u32 = 1;
const STREAM_NAME: &str = "test-stream";
const TOPIC_NAME: &str = "test-topic";
const PARTITIONS_COUNT: u32 = 3;
@@ -54,8 +52,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
client
.send_messages(
- &STREAM_ID.try_into().unwrap(),
- &TOPIC_ID.try_into().unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&Partitioning::partition_id(PARTITION_ID),
&mut messages,
)
@@ -65,8 +63,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 2. Poll messages and validate the headers
let polled_messages = client
.poll_messages(
- &STREAM_ID.try_into().unwrap(),
- &TOPIC_ID.try_into().unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
&Consumer::default(),
&PollingStrategy::offset(0),
@@ -113,19 +111,18 @@ pub async fn run(client_factory: &dyn ClientFactory) {
async fn init_system(client: &IggyClient) {
// 1. Create the stream
client
- .create_stream(STREAM_NAME, Some(STREAM_ID))
+ .create_stream(STREAM_NAME)
.await
.unwrap();
// 2. Create the topic
client
.create_topic(
- &STREAM_ID.try_into().unwrap(),
+ &STREAM_NAME.try_into().unwrap(),
TOPIC_NAME,
PARTITIONS_COUNT,
Default::default(),
None,
- None,
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -135,7 +132,7 @@ async fn init_system(client: &IggyClient) {
async fn cleanup_system(client: &IggyClient) {
client
- .delete_stream(&STREAM_ID.try_into().unwrap())
+ .delete_stream(&STREAM_NAME.try_into().unwrap())
.await
.unwrap();
}
diff --git a/core/integration/tests/server/scenarios/delete_segments_scenario.rs b/core/integration/tests/server/scenarios/delete_segments_scenario.rs
index 45a21979..6fa4a38f 100644
--- a/core/integration/tests/server/scenarios/delete_segments_scenario.rs
+++ b/core/integration/tests/server/scenarios/delete_segments_scenario.rs
@@ -37,18 +37,17 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
.unwrap();
client
- .create_stream(STREAM_NAME, Some(STREAM_ID))
+ .create_stream(STREAM_NAME)
.await
.unwrap();
client
.create_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
TOPIC_NAME,
PARTITION_ID,
CompressionAlgorithm::None,
None,
- Some(TOPIC_ID),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -68,8 +67,8 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
let mut messages = vec![message];
client
.send_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&Partitioning::partition_id(PARTITION_ID),
&mut messages,
)
@@ -106,8 +105,8 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
let segments_to_keep = 2u32;
let result = client
.delete_segments(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
PARTITION_ID,
segments_to_keep,
)
@@ -148,8 +147,8 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
// Test edge cases
let result = client
.delete_segments(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
PARTITION_ID,
0,
)
@@ -163,7 +162,7 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
let result = client
.delete_segments(
&Identifier::numeric(999).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
PARTITION_ID,
1,
)
@@ -175,7 +174,7 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
let result = client
.delete_segments(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
&Identifier::numeric(999).unwrap(),
PARTITION_ID,
1,
@@ -188,8 +187,8 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
let result = client
.delete_segments(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
999,
1,
)
@@ -201,14 +200,14 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
client
.delete_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
)
.await
.unwrap();
client
- .delete_stream(&Identifier::numeric(STREAM_ID).unwrap())
+ .delete_stream(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap();
}
diff --git a/core/integration/tests/server/scenarios/encryption_scenario.rs b/core/integration/tests/server/scenarios/encryption_scenario.rs
index d8b884a4..1239583c 100644
--- a/core/integration/tests/server/scenarios/encryption_scenario.rs
+++ b/core/integration/tests/server/scenarios/encryption_scenario.rs
@@ -78,7 +78,7 @@ async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
let partition_count = 1;
client
- .create_stream("test-stream-api", Some(stream_id))
+ .create_stream("test-stream-api")
.await
.unwrap();
client
@@ -88,10 +88,8 @@ async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
partition_count,
CompressionAlgorithm::default(),
None,
- Some(topic_id),
IggyExpiry::NeverExpire,
- MaxTopicSize::ServerDefault,
- )
+ MaxTopicSize::ServerDefault)
.await
.unwrap();
diff --git a/core/integration/tests/server/scenarios/message_headers_scenario.rs b/core/integration/tests/server/scenarios/message_headers_scenario.rs
index dda14682..21335618 100644
--- a/core/integration/tests/server/scenarios/message_headers_scenario.rs
+++ b/core/integration/tests/server/scenarios/message_headers_scenario.rs
@@ -17,7 +17,7 @@
*/
use crate::server::scenarios::{
- MESSAGES_COUNT, PARTITION_ID, PARTITIONS_COUNT, STREAM_ID, STREAM_NAME, TOPIC_ID, TOPIC_NAME,
+ MESSAGES_COUNT, PARTITION_ID, PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME,
cleanup, create_client,
};
use bytes::Bytes;
@@ -49,8 +49,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
client
.send_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&Partitioning::partition_id(PARTITION_ID),
&mut messages,
)
@@ -61,8 +61,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let consumer = Consumer::default();
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
&consumer,
&PollingStrategy::offset(0),
@@ -109,19 +109,18 @@ pub async fn run(client_factory: &dyn ClientFactory) {
async fn init_system(client: &IggyClient) {
// 1. Create the stream
client
- .create_stream(STREAM_NAME, Some(STREAM_ID))
+ .create_stream(STREAM_NAME)
.await
.unwrap();
// 2. Create the topic
client
.create_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
TOPIC_NAME,
PARTITIONS_COUNT,
CompressionAlgorithm::default(),
None,
- Some(TOPIC_ID),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
diff --git a/core/integration/tests/server/scenarios/message_size_scenario.rs b/core/integration/tests/server/scenarios/message_size_scenario.rs
index e1e320d1..791cd2ce 100644
--- a/core/integration/tests/server/scenarios/message_size_scenario.rs
+++ b/core/integration/tests/server/scenarios/message_size_scenario.rs
@@ -110,7 +110,7 @@ async fn assert_message_count(client: &IggyClient, expected_count: u32) {
async fn init_system(client: &IggyClient) {
// 1. Create the stream
client
- .create_stream(STREAM_NAME, Some(STREAM_ID))
+ .create_stream(STREAM_NAME)
.await
.unwrap();
@@ -122,9 +122,8 @@ async fn init_system(client: &IggyClient) {
PARTITIONS_COUNT,
Default::default(),
None,
- None,
IggyExpiry::NeverExpire,
- MaxTopicSize::ServerDefault,
+ MaxTopicSize::ServerDefault
)
.await
.unwrap();
diff --git a/core/integration/tests/server/scenarios/mod.rs b/core/integration/tests/server/scenarios/mod.rs
index b37835ea..50029a8d 100644
--- a/core/integration/tests/server/scenarios/mod.rs
+++ b/core/integration/tests/server/scenarios/mod.rs
@@ -33,13 +33,10 @@ pub mod user_scenario;
use iggy::prelude::*;
use integration::test_server::{ClientFactory, delete_user};
-const STREAM_ID: u32 = 1;
-const TOPIC_ID: u32 = 1;
-const PARTITION_ID: u32 = 1;
+const PARTITION_ID: u32 = 0;
const STREAM_NAME: &str = "test-stream";
const TOPIC_NAME: &str = "test-topic";
const PARTITIONS_COUNT: u32 = 3;
-const CONSUMER_GROUP_ID: u32 = 10;
const CONSUMER_GROUP_NAME: &str = "test-consumer-group";
const USERNAME_1: &str = "user1";
const USERNAME_2: &str = "user2";
@@ -56,9 +53,9 @@ async fn create_client(client_factory: &dyn ClientFactory) -> IggyClient {
async fn get_consumer_group(client: &IggyClient) -> ConsumerGroupDetails {
client
.get_consumer_group(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
- &Identifier::numeric(CONSUMER_GROUP_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
)
.await
.unwrap()
@@ -68,9 +65,9 @@ async fn get_consumer_group(client: &IggyClient) -> ConsumerGroupDetails {
async fn join_consumer_group(client: &IggyClient) {
client
.join_consumer_group(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
&Identifier::numeric(TOPIC_ID).unwrap(),
- &Identifier::numeric(CONSUMER_GROUP_ID).unwrap(),
+ &Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
)
.await
.unwrap();
@@ -79,9 +76,9 @@ async fn join_consumer_group(client: &IggyClient) {
async fn leave_consumer_group(client: &IggyClient) {
client
.leave_consumer_group(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
&Identifier::numeric(TOPIC_ID).unwrap(),
- &Identifier::numeric(CONSUMER_GROUP_ID).unwrap(),
+ &Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
)
.await
.unwrap();
@@ -94,7 +91,7 @@ async fn cleanup(system_client: &IggyClient, delete_users: bool) {
delete_user(system_client, USERNAME_3).await;
}
system_client
- .delete_stream(&Identifier::numeric(STREAM_ID).unwrap())
+ .delete_stream(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap();
}
diff --git a/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs b/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs
index cffaa9d5..c816b2da 100644
--- a/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs
+++ b/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs
@@ -137,7 +137,6 @@ async fn create_topic_assert_empty(client: &IggyClient, stream_name: &str, topic
PARTITIONS_COUNT,
CompressionAlgorithm::default(),
None,
- None,
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -150,7 +149,7 @@ async fn create_topic_assert_empty(client: &IggyClient, stream_name: &str, topic
async fn create_stream_assert_empty(client: &IggyClient, stream_name: &str) {
// 1. Create stream
- client.create_stream(stream_name, None).await.unwrap();
+ client.create_stream(stream_name).await.unwrap();
// 2. Validate stream size and number of messages
validate_stream(client, stream_name, 0, 0).await;
diff --git a/core/integration/tests/server/scenarios/system_scenario.rs b/core/integration/tests/server/scenarios/system_scenario.rs
index 305c0f5a..5e9967fc 100644
--- a/core/integration/tests/server/scenarios/system_scenario.rs
+++ b/core/integration/tests/server/scenarios/system_scenario.rs
@@ -17,8 +17,8 @@
*/
use crate::server::scenarios::{
- CONSUMER_GROUP_ID, CONSUMER_GROUP_NAME, CONSUMER_ID, CONSUMER_KIND, MESSAGES_COUNT,
- PARTITION_ID, PARTITIONS_COUNT, STREAM_ID, STREAM_NAME, TOPIC_ID, TOPIC_NAME,
+ CONSUMER_GROUP_NAME, CONSUMER_ID, CONSUMER_KIND, MESSAGES_COUNT,
+ PARTITION_ID, PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME,
get_consumer_group, leave_consumer_group,
};
use bytes::Bytes;
@@ -32,7 +32,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let consumer = Consumer {
kind: CONSUMER_KIND,
- id: Identifier::numeric(CONSUMER_ID).unwrap(),
+ id: Identifier::named("test-consumer").unwrap(),
};
// 0. Ping server
@@ -50,18 +50,18 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 3. Create the stream
let stream = client
- .create_stream(STREAM_NAME, Some(STREAM_ID))
+ .create_stream(STREAM_NAME)
.await
.unwrap();
- assert_eq!(stream.id, STREAM_ID);
+ let stream_id = stream.id;
assert_eq!(stream.name, STREAM_NAME);
// 4. Get streams and validate that created stream exists
let streams = client.get_streams().await.unwrap();
assert_eq!(streams.len(), 1);
let stream = streams.first().unwrap();
- assert_eq!(stream.id, STREAM_ID);
+ assert_eq!(stream.id, stream_id);
assert_eq!(stream.name, STREAM_NAME);
assert_eq!(stream.topics_count, 0);
assert_eq!(stream.size, 0);
@@ -69,12 +69,12 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 5. Get stream details by ID
let stream = client
- .get_stream(&Identifier::numeric(STREAM_ID).unwrap())
+ .get_stream(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap()
.expect("Failed to get stream");
- assert_eq!(stream.id, STREAM_ID);
+ assert_eq!(stream.id, stream_id);
assert_eq!(stream.name, STREAM_NAME);
assert_eq!(stream.topics_count, 0);
assert!(stream.topics.is_empty());
@@ -88,45 +88,44 @@ pub async fn run(client_factory: &dyn ClientFactory) {
.unwrap()
.expect("Failed to get stream");
- assert_eq!(stream.id, STREAM_ID);
+ assert_eq!(stream.id, stream_id);
assert_eq!(stream.name, STREAM_NAME);
- // 7. Try to create the stream with the same ID but the different name and validate that it fails
+ // 7. Try to create the stream with the same name and validate that it fails
let create_stream_result = client
- .create_stream(&format!("{STREAM_NAME}-2"), Some(STREAM_ID))
+ .create_stream(STREAM_NAME)
.await;
assert!(create_stream_result.is_err());
- // 8. Try to create the stream with the same name but the different ID and validate that it fails
- let create_stream_result = client.create_stream(STREAM_NAME, Some(STREAM_ID + 1)).await;
- assert!(create_stream_result.is_err());
+ // 8. Try to create the stream with the different name and validate that it succeeds
+ let create_stream_result = client.create_stream(&format!("{STREAM_NAME}-2")).await;
+ assert!(create_stream_result.is_ok());
// 9. Create the topic
let topic = client
.create_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
TOPIC_NAME,
PARTITIONS_COUNT,
Default::default(),
None,
- Some(TOPIC_ID),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
.await
.unwrap();
- assert_eq!(topic.id, TOPIC_ID);
+ let topic_id = topic.id;
assert_eq!(topic.name, TOPIC_NAME);
// 10. Get topics and validate that created topic exists
let topics = client
- .get_topics(&Identifier::numeric(STREAM_ID).unwrap())
+ .get_topics(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap();
assert_eq!(topics.len(), 1);
let topic = topics.first().unwrap();
- assert_eq!(topic.id, TOPIC_ID);
+ assert_eq!(topic.id, topic_id);
assert_eq!(topic.name, TOPIC_NAME);
assert_eq!(topic.partitions_count, PARTITIONS_COUNT);
assert_eq!(topic.compression_algorithm, CompressionAlgorithm::default());
@@ -139,13 +138,13 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 11. Get topic details by ID
let topic = client
.get_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
)
.await
.unwrap()
.expect("Failed to get topic");
- assert_eq!(topic.id, TOPIC_ID);
+ assert_eq!(topic.id, topic_id);
assert_eq!(topic.name, TOPIC_NAME);
assert_eq!(topic.partitions_count, PARTITIONS_COUNT);
assert_eq!(topic.partitions.len(), PARTITIONS_COUNT as usize);
@@ -164,22 +163,22 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 12. Get topic details by name
let topic = client
.get_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
&Identifier::named(TOPIC_NAME).unwrap(),
)
.await
.unwrap()
.expect("Failed to get topic");
- assert_eq!(topic.id, TOPIC_ID);
+ assert_eq!(topic.id, topic_id);
assert_eq!(topic.name, TOPIC_NAME);
// 13. Get stream details and validate that created topic exists
let stream = client
- .get_stream(&Identifier::numeric(STREAM_ID).unwrap())
+ .get_stream(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap()
.expect("Failed to get stream");
- assert_eq!(stream.id, STREAM_ID);
+ assert_eq!(stream.id, stream_id);
assert_eq!(stream.name, STREAM_NAME);
assert_eq!(stream.topics_count, 1);
assert_eq!(stream.topics.len(), 1);
@@ -191,30 +190,28 @@ pub async fn run(client_factory: &dyn ClientFactory) {
assert_eq!(stream_topic.size, 0);
assert_eq!(stream_topic.messages_count, 0);
- // 15. Try to create the topic with the same ID but the different name and validate that it fails
+ // 15. Try to create the topic with the same name and validate that it fails
let create_topic_result = client
.create_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &format!("{TOPIC_NAME}-2"),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ TOPIC_NAME,
PARTITIONS_COUNT,
Default::default(),
None,
- Some(TOPIC_ID),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
.await;
assert!(create_topic_result.is_err());
- // 16. Try to create the topic with the different ID but the same name and validate that it fails
+ // 16. Try to create the topic with the different name and validate that it succeeds
let create_topic_result = client
.create_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
- TOPIC_NAME,
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &format!("{TOPIC_NAME}-2"),
PARTITIONS_COUNT,
Default::default(),
None,
- Some(TOPIC_ID + 1),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -225,8 +222,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let mut messages = create_messages();
client
.send_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&Partitioning::partition_id(PARTITION_ID),
&mut messages,
)
@@ -236,8 +233,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 18. Poll messages from the specific partition in topic
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
&consumer,
&PollingStrategy::offset(0),
@@ -260,8 +257,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let start_offset = (i * batch_length) as u64;
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
&consumer,
&PollingStrategy::offset(start_offset),
@@ -281,13 +278,13 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 20. Get topic details and validate the partition details
let topic = client
.get_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
&Identifier::named(TOPIC_NAME).unwrap(),
)
.await
.unwrap()
.expect("Failed to get topic");
- assert_eq!(topic.id, TOPIC_ID);
+ assert_eq!(topic.id, topic_id);
assert_eq!(topic.name, TOPIC_NAME);
assert_eq!(topic.partitions_count, PARTITIONS_COUNT);
assert_eq!(topic.partitions.len(), PARTITIONS_COUNT as usize);
@@ -303,8 +300,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 21. Ensure that messages do not exist in the second partition in the same topic
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID + 1),
&consumer,
&PollingStrategy::offset(0),
@@ -319,8 +316,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let offset = client
.get_consumer_offset(
&consumer,
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
)
.await
@@ -332,8 +329,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
client
.store_consumer_offset(
&consumer,
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
stored_offset,
)
@@ -344,8 +341,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let offset = client
.get_consumer_offset(
&consumer,
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
)
.await
@@ -359,8 +356,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
client
.delete_consumer_offset(
&consumer,
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
)
.await
@@ -370,8 +367,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let offset = client
.get_consumer_offset(
&consumer,
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
)
.await
@@ -382,8 +379,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
client
.store_consumer_offset(
&consumer,
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
stored_offset,
)
@@ -394,8 +391,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let messages_count = 10;
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
&consumer,
&PollingStrategy::next(),
@@ -415,8 +412,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let offset = client
.get_consumer_offset(
&consumer,
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
)
.await
@@ -429,8 +426,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 29. Get the consumer groups and validate that there are no groups
let consumer_groups = client
.get_consumer_groups(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
)
.await
.unwrap();
@@ -440,44 +437,43 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 30. Create the consumer group
let consumer_group = client
.create_consumer_group(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
CONSUMER_GROUP_NAME,
- Some(CONSUMER_GROUP_ID),
)
.await
.unwrap();
- assert_eq!(consumer_group.id, CONSUMER_GROUP_ID);
+ let consumer_group_id = consumer_group.id;
assert_eq!(consumer_group.name, CONSUMER_GROUP_NAME);
// 31. Get the consumer groups and validate that there is one group
let consumer_groups = client
.get_consumer_groups(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
)
.await
.unwrap();
assert_eq!(consumer_groups.len(), 1);
let consumer_group = consumer_groups.first().unwrap();
- assert_eq!(consumer_group.id, CONSUMER_GROUP_ID);
+ assert_eq!(consumer_group.id, consumer_group_id);
assert_eq!(consumer_group.partitions_count, PARTITIONS_COUNT);
assert_eq!(consumer_group.members_count, 0);
// 32. Get the consumer group details
let consumer_group = client
.get_consumer_group(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
- &Identifier::numeric(CONSUMER_GROUP_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
)
.await
.unwrap()
.expect("Failed to get consumer group");
- assert_eq!(consumer_group.id, CONSUMER_GROUP_ID);
+ assert_eq!(consumer_group.id, consumer_group_id);
assert_eq!(consumer_group.partitions_count, PARTITIONS_COUNT);
assert_eq!(consumer_group.members_count, 0);
assert!(consumer_group.members.is_empty());
@@ -485,16 +481,16 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 33. Join the consumer group and then leave it if the feature is available
let result = client
.join_consumer_group(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
- &Identifier::numeric(CONSUMER_GROUP_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
)
.await;
match result {
Ok(_) => {
let consumer_group = get_consumer_group(&client).await;
- assert_eq!(consumer_group.id, CONSUMER_GROUP_ID);
+ assert_eq!(consumer_group.id, consumer_group_id);
assert_eq!(consumer_group.partitions_count, PARTITIONS_COUNT);
assert_eq!(consumer_group.name, CONSUMER_GROUP_NAME);
assert_eq!(consumer_group.members_count, 1);
@@ -507,9 +503,9 @@ pub async fn run(client_factory: &dyn ClientFactory) {
assert_eq!(me.consumer_groups_count, 1);
assert_eq!(me.consumer_groups.len(), 1);
let consumer_group = &me.consumer_groups[0];
- assert_eq!(consumer_group.stream_id, STREAM_ID);
- assert_eq!(consumer_group.topic_id, TOPIC_ID);
- assert_eq!(consumer_group.group_id, CONSUMER_GROUP_ID);
+ assert_eq!(consumer_group.stream_id, stream_id);
+ assert_eq!(consumer_group.topic_id, topic_id);
+ assert_eq!(consumer_group.group_id, consumer_group_id);
leave_consumer_group(&client).await;
@@ -543,9 +539,9 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 35. Delete the consumer group
client
.delete_consumer_group(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
- &Identifier::numeric(CONSUMER_GROUP_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
)
.await
.unwrap();
@@ -553,8 +549,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 36. Create new partitions and validate that the number of partitions is increased
client
.create_partitions(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
PARTITIONS_COUNT,
)
.await
@@ -562,8 +558,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let topic = client
.get_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
)
.await
.unwrap()
@@ -574,8 +570,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 37. Delete the partitions and validate that the number of partitions is decreased
client
.delete_partitions(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
PARTITIONS_COUNT,
)
.await
@@ -583,8 +579,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let topic = client
.get_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
)
.await
.unwrap()
@@ -601,8 +597,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
client
.update_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&updated_topic_name,
CompressionAlgorithm::Gzip,
Some(updated_replication_factor),
@@ -614,8 +610,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let updated_topic = client
.get_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
)
.await
.unwrap()
@@ -636,16 +632,16 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 39. Purge the existing topic and ensure it has no messages
client
.purge_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
)
.await
.unwrap();
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
&consumer,
&PollingStrategy::offset(0),
@@ -662,14 +658,14 @@ pub async fn run(client_factory: &dyn ClientFactory) {
client
.update_stream(
- &Identifier::numeric(STREAM_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
&updated_stream_name,
)
.await
.unwrap();
let updated_stream = client
- .get_stream(&Identifier::numeric(STREAM_ID).unwrap())
+ .get_stream(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap()
.expect("Failed to get stream");
@@ -680,8 +676,8 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let mut messages = create_messages();
client
.send_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&Partitioning::partition_id(PARTITION_ID),
&mut messages,
)
@@ -689,14 +685,14 @@ pub async fn run(client_factory: &dyn ClientFactory) {
.unwrap();
client
- .purge_stream(&Identifier::numeric(STREAM_ID).unwrap())
+ .purge_stream(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap();
let polled_messages = client
.poll_messages(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(PARTITION_ID),
&consumer,
&PollingStrategy::offset(0),
@@ -711,42 +707,39 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 42. Delete the existing topic and ensure it doesn't exist anymore
client
.delete_topic(
- &Identifier::numeric(STREAM_ID).unwrap(),
- &Identifier::numeric(TOPIC_ID).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
)
.await
.unwrap();
let topics = client
- .get_topics(&Identifier::numeric(STREAM_ID).unwrap())
+ .get_topics(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap();
assert!(topics.is_empty());
// 43. Create the stream with automatically generated ID on the server
let stream_name = format!("{STREAM_NAME}-auto");
- let stream_id = STREAM_ID + 1;
- client.create_stream(&stream_name, None).await.unwrap();
+ let stream = client.create_stream(&stream_name).await.unwrap();
let stream = client
- .get_stream(&Identifier::numeric(stream_id).unwrap())
+ .get_stream(&Identifier::named(&stream_name).unwrap())
.await
.unwrap()
.expect("Failed to get stream");
- assert_eq!(stream.id, stream_id);
+ let stream_id = stream.id;
assert_eq!(stream.name, stream_name);
// 44. Create the topic with automatically generated ID on the server
let topic_name = format!("{TOPIC_NAME}-auto");
- let topic_id = 1;
- client
+ let topic = client
.create_topic(
- &Identifier::numeric(stream_id).unwrap(),
+ &Identifier::named(&stream_name).unwrap(),
&topic_name,
PARTITIONS_COUNT,
CompressionAlgorithm::default(),
None,
- None,
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -755,14 +748,14 @@ pub async fn run(client_factory: &dyn ClientFactory) {
let topic = client
.get_topic(
- &Identifier::numeric(stream_id).unwrap(),
- &Identifier::numeric(topic_id).unwrap(),
+ &Identifier::named(&stream_name).unwrap(),
+ &Identifier::named(&topic_name).unwrap(),
)
.await
.unwrap()
.expect("Failed to get topic");
- assert_eq!(topic.id, topic_id);
+ let topic_id = topic.id;
assert_eq!(topic.name, topic_name);
// 45. Delete the existing streams and ensure there's no streams left
diff --git a/core/integration/tests/server/scenarios/tcp_tls_scenario.rs b/core/integration/tests/server/scenarios/tcp_tls_scenario.rs
index 315cf5ea..947fcf08 100644
--- a/core/integration/tests/server/scenarios/tcp_tls_scenario.rs
+++ b/core/integration/tests/server/scenarios/tcp_tls_scenario.rs
@@ -24,29 +24,26 @@ pub async fn run(client: &IggyClient) {
login_root(client).await;
let stream_name = "test-tls-stream";
- let stream_id = 1;
client
- .create_stream(stream_name, Some(stream_id))
+ .create_stream(stream_name)
.await
.unwrap();
let stream = client
- .get_stream(&Identifier::numeric(stream_id).unwrap())
+ .get_stream(&Identifier::named(stream_name).unwrap())
.await
.unwrap()
.unwrap();
assert_eq!(stream.name, stream_name);
let topic_name = "test-tls-topic";
- let topic_id = 1;
client
.create_topic(
- &Identifier::numeric(stream_id).unwrap(),
+ &Identifier::named(stream_name).unwrap(),
topic_name,
1,
CompressionAlgorithm::default(),
None,
- Some(topic_id),
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
@@ -63,8 +60,8 @@ pub async fn run(client: &IggyClient) {
client
.send_messages(
- &Identifier::numeric(stream_id).unwrap(),
- &Identifier::numeric(topic_id).unwrap(),
+ &Identifier::named(stream_name).unwrap(),
+ &Identifier::named(topic_name).unwrap(),
&Partitioning::partition_id(1),
&mut messages,
)
@@ -73,8 +70,8 @@ pub async fn run(client: &IggyClient) {
let polled_messages = client
.poll_messages(
- &Identifier::numeric(stream_id).unwrap(),
- &Identifier::numeric(topic_id).unwrap(),
+ &Identifier::named(stream_name).unwrap(),
+ &Identifier::named(topic_name).unwrap(),
Some(1),
&Consumer::default(),
&PollingStrategy::offset(0),
@@ -88,7 +85,7 @@ pub async fn run(client: &IggyClient) {
assert_eq!(polled_messages.messages[0].payload.as_ref(), b"Hello TLS!");
client
- .delete_stream(&Identifier::numeric(stream_id).unwrap())
+ .delete_stream(&Identifier::named(stream_name).unwrap())
.await
.unwrap();
diff --git a/core/integration/tests/state/file.rs b/core/integration/tests/state/file.rs
index 476e17f4..b830cd53 100644
--- a/core/integration/tests/state/file.rs
+++ b/core/integration/tests/state/file.rs
@@ -120,7 +120,6 @@ async fn should_apply_multiple_entries() {
let create_stream = EntryCommand::CreateStream(CreateStreamWithId {
stream_id,
command: CreateStream {
- stream_id: Some(stream_id),
name: "test".to_string(),
},
});
diff --git a/core/integration/tests/state/system.rs b/core/integration/tests/state/system.rs
index 8913f743..63079d57 100644
--- a/core/integration/tests/state/system.rs
+++ b/core/integration/tests/state/system.rs
@@ -55,19 +55,16 @@ async fn should_be_initialized_based_on_state_entries() {
let stream1_id = 1;
let create_stream1 = CreateStream {
- stream_id: Some(stream1_id),
name: "stream1".to_string(),
};
let create_stream1_clone = CreateStream {
- stream_id: Some(stream1_id),
name: "stream1".to_string(),
};
let topic1_id = 1;
let create_topic1 = CreateTopic {
stream_id: stream1_id.try_into().unwrap(),
- topic_id: Some(topic1_id),
partitions_count: 1,
compression_algorithm: Default::default(),
message_expiry: Default::default(),
@@ -78,7 +75,6 @@ async fn should_be_initialized_based_on_state_entries() {
let create_topic1_clone = CreateTopic {
stream_id: stream1_id.try_into().unwrap(),
- topic_id: Some(topic1_id),
partitions_count: 1,
compression_algorithm: Default::default(),
message_expiry: Default::default(),
@@ -89,14 +85,12 @@ async fn should_be_initialized_based_on_state_entries() {
let stream2_id = 2;
let create_stream2 = CreateStream {
- stream_id: Some(stream2_id),
name: "stream2".to_string(),
};
let topic2_id = 2;
let create_topic2 = CreateTopic {
stream_id: stream2_id.try_into().unwrap(),
- topic_id: Some(topic2_id),
partitions_count: 1,
compression_algorithm: Default::default(),
message_expiry: Default::default(),
@@ -135,14 +129,12 @@ async fn should_be_initialized_based_on_state_entries() {
let create_consumer_group = CreateConsumerGroup {
stream_id: stream1_id.try_into().unwrap(),
topic_id: topic1_id.try_into().unwrap(),
- group_id: Some(group_id),
name: "test".to_string(),
};
let create_consumer_group_clone = CreateConsumerGroup {
stream_id: stream1_id.try_into().unwrap(),
topic_id: topic1_id.try_into().unwrap(),
- group_id: Some(group_id),
name: "test".to_string(),
};
@@ -250,17 +242,17 @@ async fn should_be_initialized_based_on_state_entries() {
assert_eq!(system.streams.len(), 1);
let mut stream = system
.streams
- .remove(&create_stream1_clone.stream_id.unwrap())
+ .remove(&stream1_id)
.unwrap();
- assert_eq!(stream.id, create_stream1_clone.stream_id.unwrap());
+ assert_eq!(stream.id, stream1_id);
assert_eq!(stream.name, create_stream1_clone.name);
assert_eq!(stream.topics.len(), 1);
let mut topic = stream
.topics
- .remove(&create_topic1_clone.topic_id.unwrap())
+ .remove(&topic1_id)
.unwrap();
- assert_eq!(topic.id, create_topic1_clone.topic_id.unwrap());
+ assert_eq!(topic.id, topic1_id);
assert_eq!(topic.name, create_topic1_clone.name);
assert_eq!(topic.partitions.len(), 3);
@@ -269,7 +261,7 @@ async fn should_be_initialized_based_on_state_entries() {
assert_eq!(
consumer_group.id,
- create_consumer_group_clone.group_id.unwrap()
+ group_id
);
assert_eq!(consumer_group.name, create_consumer_group_clone.name);
}
diff --git a/core/sdk/src/client_wrappers/binary_consumer_group_client.rs b/core/sdk/src/client_wrappers/binary_consumer_group_client.rs
index 18117588..4c2c9ecd 100644
--- a/core/sdk/src/client_wrappers/binary_consumer_group_client.rs
+++ b/core/sdk/src/client_wrappers/binary_consumer_group_client.rs
@@ -72,27 +72,26 @@ impl ConsumerGroupClient for ClientWrapper {
stream_id: &Identifier,
topic_id: &Identifier,
name: &str,
- group_id: Option<u32>,
) -> Result<ConsumerGroupDetails, IggyError> {
match self {
ClientWrapper::Iggy(client) => {
client
- .create_consumer_group(stream_id, topic_id, name, group_id)
+ .create_consumer_group(stream_id, topic_id, name)
.await
}
ClientWrapper::Http(client) => {
client
- .create_consumer_group(stream_id, topic_id, name, group_id)
+ .create_consumer_group(stream_id, topic_id, name)
.await
}
ClientWrapper::Tcp(client) => {
client
- .create_consumer_group(stream_id, topic_id, name, group_id)
+ .create_consumer_group(stream_id, topic_id, name)
.await
}
ClientWrapper::Quic(client) => {
client
- .create_consumer_group(stream_id, topic_id, name, group_id)
+ .create_consumer_group(stream_id, topic_id, name)
.await
}
}
diff --git a/core/sdk/src/client_wrappers/binary_stream_client.rs b/core/sdk/src/client_wrappers/binary_stream_client.rs
index 93d785dd..e1ad849a 100644
--- a/core/sdk/src/client_wrappers/binary_stream_client.rs
+++ b/core/sdk/src/client_wrappers/binary_stream_client.rs
@@ -44,13 +44,12 @@ impl StreamClient for ClientWrapper {
async fn create_stream(
&self,
name: &str,
- stream_id: Option<u32>,
) -> Result<StreamDetails, IggyError> {
match self {
- ClientWrapper::Iggy(client) => client.create_stream(name, stream_id).await,
- ClientWrapper::Http(client) => client.create_stream(name, stream_id).await,
- ClientWrapper::Tcp(client) => client.create_stream(name, stream_id).await,
- ClientWrapper::Quic(client) => client.create_stream(name, stream_id).await,
+ ClientWrapper::Iggy(client) => client.create_stream(name).await,
+ ClientWrapper::Http(client) => client.create_stream(name).await,
+ ClientWrapper::Tcp(client) => client.create_stream(name).await,
+ ClientWrapper::Quic(client) => client.create_stream(name).await,
}
}
diff --git a/core/sdk/src/client_wrappers/binary_topic_client.rs b/core/sdk/src/client_wrappers/binary_topic_client.rs
index c9685bdb..0f877226 100644
--- a/core/sdk/src/client_wrappers/binary_topic_client.rs
+++ b/core/sdk/src/client_wrappers/binary_topic_client.rs
@@ -54,7 +54,6 @@ impl TopicClient for ClientWrapper {
partitions_count: u32,
compression_algorithm: CompressionAlgorithm,
replication_factor: Option<u8>,
- topic_id: Option<u32>,
message_expiry: IggyExpiry,
max_topic_size: MaxTopicSize,
) -> Result<TopicDetails, IggyError> {
@@ -67,7 +66,6 @@ impl TopicClient for ClientWrapper {
partitions_count,
compression_algorithm,
replication_factor,
- topic_id,
message_expiry,
max_topic_size,
)
@@ -81,7 +79,6 @@ impl TopicClient for ClientWrapper {
partitions_count,
compression_algorithm,
replication_factor,
- topic_id,
message_expiry,
max_topic_size,
)
@@ -95,7 +92,6 @@ impl TopicClient for ClientWrapper {
partitions_count,
compression_algorithm,
replication_factor,
- topic_id,
message_expiry,
max_topic_size,
)
@@ -109,7 +105,6 @@ impl TopicClient for ClientWrapper {
partitions_count,
compression_algorithm,
replication_factor,
- topic_id,
message_expiry,
max_topic_size,
)
diff --git a/core/sdk/src/clients/binary_consumer_group.rs b/core/sdk/src/clients/binary_consumer_group.rs
index b415e42f..4209dba3 100644
--- a/core/sdk/src/clients/binary_consumer_group.rs
+++ b/core/sdk/src/clients/binary_consumer_group.rs
@@ -56,12 +56,11 @@ impl ConsumerGroupClient for IggyClient {
stream_id: &Identifier,
topic_id: &Identifier,
name: &str,
- group_id: Option<u32>,
) -> Result<ConsumerGroupDetails, IggyError> {
self.client
.read()
.await
- .create_consumer_group(stream_id, topic_id, name, group_id)
+ .create_consumer_group(stream_id, topic_id, name)
.await
}
diff --git a/core/sdk/src/clients/binary_streams.rs b/core/sdk/src/clients/binary_streams.rs
index 4001ec4f..74cf30ac 100644
--- a/core/sdk/src/clients/binary_streams.rs
+++ b/core/sdk/src/clients/binary_streams.rs
@@ -35,12 +35,11 @@ impl StreamClient for IggyClient {
async fn create_stream(
&self,
name: &str,
- stream_id: Option<u32>,
) -> Result<StreamDetails, IggyError> {
self.client
.read()
.await
- .create_stream(name, stream_id)
+ .create_stream(name)
.await
}
diff --git a/core/sdk/src/clients/binary_topics.rs b/core/sdk/src/clients/binary_topics.rs
index 225336ea..28c2b346 100644
--- a/core/sdk/src/clients/binary_topics.rs
+++ b/core/sdk/src/clients/binary_topics.rs
@@ -49,7 +49,6 @@ impl TopicClient for IggyClient {
partitions_count: u32,
compression_algorithm: CompressionAlgorithm,
replication_factor: Option<u8>,
- topic_id: Option<u32>,
message_expiry: IggyExpiry,
max_topic_size: MaxTopicSize,
) -> Result<TopicDetails, IggyError> {
@@ -62,7 +61,6 @@ impl TopicClient for IggyClient {
partitions_count,
compression_algorithm,
replication_factor,
- topic_id,
message_expiry,
max_topic_size,
)
diff --git a/core/sdk/src/clients/consumer.rs b/core/sdk/src/clients/consumer.rs
index fbc698ad..66975eae 100644
--- a/core/sdk/src/clients/consumer.rs
+++ b/core/sdk/src/clients/consumer.rs
@@ -813,7 +813,7 @@ impl IggyConsumer {
}
let client = client.read().await;
- let (name, id) = match consumer.id.kind {
+ let (name, _id) = match consumer.id.kind {
IdKind::Numeric => (consumer_name.to_owned(), Some(consumer.id.get_u32_value()?)),
IdKind::String => (consumer.id.get_string_value()?, None),
};
@@ -839,7 +839,7 @@ impl IggyConsumer {
"Creating consumer group: {consumer_group_id} for topic: {topic_id}, stream: {stream_id}"
);
match client
- .create_consumer_group(&stream_id, &topic_id, &name, id)
+ .create_consumer_group(&stream_id, &topic_id, &name)
.await
{
Ok(_) => {}
diff --git a/core/sdk/src/clients/producer.rs b/core/sdk/src/clients/producer.rs
index 6cd7f03a..0a6ec018 100644
--- a/core/sdk/src/clients/producer.rs
+++ b/core/sdk/src/clients/producer.rs
@@ -92,7 +92,7 @@ impl ProducerCore {
return Err(IggyError::StreamNameNotFound(self.stream_name.clone()));
}
- let (name, id) = match stream_id.kind {
+ let (name, _id) = match stream_id.kind {
IdKind::Numeric => (
self.stream_name.to_owned(),
Some(self.stream_id.get_u32_value()?),
@@ -100,7 +100,7 @@ impl ProducerCore {
IdKind::String => (self.stream_id.get_string_value()?, None),
};
info!("Creating stream: {name}");
- client.create_stream(&name, id).await?;
+ client.create_stream(&name).await?;
}
if client.get_topic(&stream_id, &topic_id).await?.is_none() {
@@ -112,7 +112,7 @@ impl ProducerCore {
));
}
- let (name, id) = match self.topic_id.kind {
+ let (name, _id) = match self.topic_id.kind {
IdKind::Numeric => (
self.topic_name.to_owned(),
Some(self.topic_id.get_u32_value()?),
@@ -127,7 +127,6 @@ impl ProducerCore {
self.topic_partitions_count,
CompressionAlgorithm::None,
self.topic_replication_factor,
- id,
self.topic_message_expiry,
self.topic_max_size,
)
diff --git a/core/sdk/src/http/binary_consumer_groups.rs b/core/sdk/src/http/binary_consumer_groups.rs
index f27b96c2..389c1aa9 100644
--- a/core/sdk/src/http/binary_consumer_groups.rs
+++ b/core/sdk/src/http/binary_consumer_groups.rs
@@ -75,7 +75,6 @@ impl ConsumerGroupClient for HttpClient {
stream_id: &Identifier,
topic_id: &Identifier,
name: &str,
- group_id: Option<u32>,
) -> Result<ConsumerGroupDetails, IggyError> {
let response = self
.post(
@@ -84,7 +83,6 @@ impl ConsumerGroupClient for HttpClient {
stream_id: stream_id.clone(),
topic_id: topic_id.clone(),
name: name.to_string(),
- group_id,
},
)
.await?;
diff --git a/core/sdk/src/http/binary_streams.rs b/core/sdk/src/http/binary_streams.rs
index e39791cf..fe6199de 100644
--- a/core/sdk/src/http/binary_streams.rs
+++ b/core/sdk/src/http/binary_streams.rs
@@ -59,14 +59,12 @@ impl StreamClient for HttpClient {
async fn create_stream(
&self,
name: &str,
- stream_id: Option<u32>,
) -> Result<StreamDetails, IggyError> {
let response = self
.post(
PATH,
&CreateStream {
name: name.to_string(),
- stream_id,
},
)
.await?;
diff --git a/core/sdk/src/http/binary_topics.rs b/core/sdk/src/http/binary_topics.rs
index 97633b94..7ac71d2a 100644
--- a/core/sdk/src/http/binary_topics.rs
+++ b/core/sdk/src/http/binary_topics.rs
@@ -69,7 +69,6 @@ impl TopicClient for HttpClient {
partitions_count: u32,
compression_algorithm: CompressionAlgorithm,
replication_factor: Option<u8>,
- topic_id: Option<u32>,
message_expiry: IggyExpiry,
max_topic_size: MaxTopicSize,
) -> Result<TopicDetails, IggyError> {
@@ -82,7 +81,6 @@ impl TopicClient for HttpClient {
partitions_count,
compression_algorithm,
replication_factor,
- topic_id,
message_expiry,
max_topic_size,
},
diff --git a/core/sdk/src/stream_builder/build/build_stream_topic.rs b/core/sdk/src/stream_builder/build/build_stream_topic.rs
index 2a8e0e73..a3dd09b0 100644
--- a/core/sdk/src/stream_builder/build/build_stream_topic.rs
+++ b/core/sdk/src/stream_builder/build/build_stream_topic.rs
@@ -60,9 +60,9 @@ pub(crate) async fn build_iggy_stream_topic_if_not_exists(
return Ok(());
}
- let (name, id) = extract_name_id_from_identifier(stream_id, stream_name)?;
+ let (name, _id) = extract_name_id_from_identifier(stream_id, stream_name)?;
trace!("Creating stream: {name}");
- client.create_stream(&name, id).await?;
+ client.create_stream(&name).await?;
}
trace!("Check if topic exists.");
@@ -85,7 +85,7 @@ pub(crate) async fn build_iggy_stream_topic_if_not_exists(
let topic_partitions_count = config.partitions_count();
let topic_replication_factor = config.replication_factor();
- let (name, id) = extract_name_id_from_identifier(topic_id, topic_name)?;
+ let (name, _id) = extract_name_id_from_identifier(topic_id, topic_name)?;
trace!("Create topic: {name} for stream: {}", stream_name);
client
.create_topic(
@@ -94,7 +94,6 @@ pub(crate) async fn build_iggy_stream_topic_if_not_exists(
topic_partitions_count,
CompressionAlgorithm::None,
topic_replication_factor,
- id,
IggyExpiry::ServerDefault,
MaxTopicSize::ServerDefault,
)
diff --git a/core/server/src/binary/handlers/streams/create_stream_handler.rs b/core/server/src/binary/handlers/streams/create_stream_handler.rs
index 2cab0ab2..21d24ff6 100644
--- a/core/server/src/binary/handlers/streams/create_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/create_stream_handler.rs
@@ -48,9 +48,8 @@ impl ServerCommandHandler for CreateStream {
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
- let stream_id = self.stream_id;
let stream = shard
- .create_stream2(session, stream_id, self.name.clone())
+ .create_stream2(session, self.name.clone())
.await?;
let created_stream_id = stream.id();
shard_info!(
@@ -79,7 +78,7 @@ impl ServerCommandHandler for CreateStream {
.await
.with_error_context(|error| {
format!(
- "{COMPONENT} (error: {error}) - failed to apply create stream for id: {stream_id:?}, session: {session}"
+ "{COMPONENT} (error: {error}) - failed to apply create stream for id: {created_stream_id}, session: {session}"
)
})?;
sender.send_ok_response(&response).await?;
diff --git a/core/server/src/http/http_shard_wrapper.rs b/core/server/src/http/http_shard_wrapper.rs
index 4ac0cddf..bafe4f60 100644
--- a/core/server/src/http/http_shard_wrapper.rs
+++ b/core/server/src/http/http_shard_wrapper.rs
@@ -134,10 +134,9 @@ impl HttpSafeShard {
pub async fn create_stream(
&self,
session: &Session,
- stream_id: Option<u32>,
name: String,
) -> Result<crate::streaming::streams::stream2::Stream, IggyError> {
- let future = SendWrapper::new(self.shard().create_stream2(session, stream_id, name));
+ let future = SendWrapper::new(self.shard().create_stream2(session, name));
future.await
}
diff --git a/core/server/src/http/streams.rs b/core/server/src/http/streams.rs
index 0e61f657..d765a91d 100644
--- a/core/server/src/http/streams.rs
+++ b/core/server/src/http/streams.rs
@@ -101,12 +101,12 @@ async fn create_stream(
// Create stream using wrapper method
let stream = state
.shard
- .create_stream(&session, command.stream_id, command.name.clone())
+ .create_stream(&session, command.name.clone())
.await
.with_error_context(|error| {
format!(
- "{COMPONENT} (error: {error}) - failed to create stream, stream ID: {:?}",
- command.stream_id
+ "{COMPONENT} (error: {error}) - failed to create stream with name: {}",
+ command.name
)
})?;
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index fb8f445b..bf0d6024 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -34,7 +34,6 @@ impl IggyShard {
pub async fn create_stream2(
&self,
session: &Session,
- _stream_id: Option<u32>,
name: String,
) -> Result<stream2::Stream, IggyError> {
self.ensure_authenticated(session)?;