This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch io_uring_tpc
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/io_uring_tpc by this push:
new 216dd6b9 feat(io_uring): fix cli tests (#2238)
216dd6b9 is described below
commit 216dd6b9c165596813c4b7d7cf45f9691e0542a3
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Mon Oct 6 14:57:27 2025 +0200
feat(io_uring): fix cli tests (#2238)
Co-authored-by: numminex <[email protected]>
---
core/cli/src/args/consumer_offset.rs | 4 +-
core/cli/src/args/message.rs | 4 +-
core/common/src/error/iggy_error.rs | 2 +
core/common/src/types/identifier/mod.rs | 7 +-
core/integration/src/test_server.rs | 99 ++++++++--
.../test_consumer_group_create_command.rs | 12 +-
.../test_consumer_group_delete_command.rs | 6 +
.../test_consumer_group_get_command.rs | 8 +-
.../test_consumer_offset_get_command.rs | 2 +-
.../test_consumer_offset_set_command.rs | 6 +-
.../tests/cli/context/test_context_applied.rs | 9 +-
.../cli/message/test_message_flush_command.rs | 3 +-
.../tests/cli/message/test_message_poll_command.rs | 47 ++---
.../message/test_message_poll_to_file_command.rs | 84 +++++---
.../tests/cli/message/test_message_send_command.rs | 102 +++++++---
.../message/test_message_send_from_file_command.rs | 139 ++++++++------
.../cli/partition/test_partition_delete_command.rs | 49 ++---
.../tests/cli/stream/test_stream_create_command.rs | 27 ++-
.../tests/cli/stream/test_stream_delete_command.rs | 16 +-
.../tests/cli/stream/test_stream_get_command.rs | 24 +--
.../tests/cli/stream/test_stream_purge_command.rs | 24 ++-
.../tests/cli/stream/test_stream_update_command.rs | 25 ++-
.../tests/cli/system/test_stats_command.rs | 9 +-
.../tests/cli/topic/test_topic_create_command.rs | 38 ++--
.../tests/cli/topic/test_topic_delete_command.rs | 30 +--
.../tests/cli/topic/test_topic_get_command.rs | 34 ++--
.../tests/cli/topic/test_topic_list_command.rs | 24 +--
.../tests/cli/topic/test_topic_purge_command.rs | 44 ++---
.../tests/cli/topic/test_topic_update_command.rs | 55 +++---
.../binary/handlers/topics/create_topic_handler.rs | 5 +-
core/server/src/bootstrap.rs | 8 +-
core/server/src/http/http_server.rs | 18 ++
core/server/src/main.rs | 2 +-
core/server/src/quic/quic_server.rs | 72 +++++--
core/server/src/shard/builder.rs | 38 ++--
core/server/src/shard/mod.rs | 212 ++++++++++++++++++---
core/server/src/shard/system/messages.rs | 5 +-
core/server/src/shard/system/partitions.rs | 18 +-
core/server/src/shard/system/snapshot/mod.rs | 58 ++++--
core/server/src/shard/system/utils.rs | 18 ++
core/server/src/shard/transmission/event.rs | 3 +-
core/server/src/slab/streams.rs | 21 +-
core/server/src/streaming/topics/topic2.rs | 4 +-
core/server/src/tcp/tcp_listener.rs | 30 +--
core/server/src/tcp/tcp_tls_listener.rs | 30 +--
45 files changed, 919 insertions(+), 556 deletions(-)
diff --git a/core/cli/src/args/consumer_offset.rs
b/core/cli/src/args/consumer_offset.rs
index bbb03158..26781b9f 100644
--- a/core/cli/src/args/consumer_offset.rs
+++ b/core/cli/src/args/consumer_offset.rs
@@ -77,7 +77,7 @@ pub(crate) struct ConsumerOffsetGetArgs {
#[arg(value_parser = clap::value_parser!(Identifier))]
pub(crate) topic_id: Identifier,
/// Partitions ID for which consumer offset is retrieved
- #[arg(value_parser = clap::value_parser!(u32).range(1..))]
+ #[arg(value_parser = clap::value_parser!(u32).range(0..))]
pub(crate) partition_id: u32,
/// Consumer kind: "consumer" for regular consumer, "consumer_group" for
consumer group
#[arg(short = 'k', long = "kind", default_value = "consumer", value_enum)]
@@ -103,7 +103,7 @@ pub(crate) struct ConsumerOffsetSetArgs {
#[arg(value_parser = clap::value_parser!(Identifier))]
pub(crate) topic_id: Identifier,
/// Partitions ID for which consumer offset is set
- #[arg(value_parser = clap::value_parser!(u32).range(1..))]
+ #[arg(value_parser = clap::value_parser!(u32).range(0..))]
pub(crate) partition_id: u32,
/// Offset to set
pub(crate) offset: u64,
diff --git a/core/cli/src/args/message.rs b/core/cli/src/args/message.rs
index 08d69348..da5e742e 100644
--- a/core/cli/src/args/message.rs
+++ b/core/cli/src/args/message.rs
@@ -146,7 +146,7 @@ pub(crate) struct PollMessagesArgs {
#[arg(value_parser = clap::value_parser!(Identifier))]
pub(crate) topic_id: Identifier,
/// Partition ID from which message will be polled
- #[arg(value_parser = clap::value_parser!(u32).range(1..))]
+ #[arg(value_parser = clap::value_parser!(u32).range(0..))]
pub(crate) partition_id: u32,
/// Number of messages to poll
#[clap(verbatim_doc_comment)]
@@ -219,7 +219,7 @@ pub(crate) struct FlushMessagesArgs {
#[arg(value_parser = clap::value_parser!(Identifier))]
pub(crate) topic_id: Identifier,
/// Partition ID for which messages will be flushed
- #[arg(value_parser = clap::value_parser!(u32).range(1..))]
+ #[arg(value_parser = clap::value_parser!(u32).range(0..))]
pub(crate) partition_id: u32,
/// fsync flushed data to disk
///
diff --git a/core/common/src/error/iggy_error.rs
b/core/common/src/error/iggy_error.rs
index 19dd41a9..92af8601 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -250,6 +250,8 @@ pub enum IggyError {
CannotReadTopics(u32) = 2017,
#[error("Invalid replication factor")]
InvalidReplicationFactor = 2018,
+ #[error("Invalid partitions count")]
+ InvalidPartitionsCount = 2019,
#[error("Cannot create partition with ID: {0} for stream with ID: {1} and
topic with ID: {2}")]
CannotCreatePartition(u32, u32, u32) = 3000,
#[error(
diff --git a/core/common/src/types/identifier/mod.rs
b/core/common/src/types/identifier/mod.rs
index 70ceb77a..bd4a5ccc 100644
--- a/core/common/src/types/identifier/mod.rs
+++ b/core/common/src/types/identifier/mod.rs
@@ -64,7 +64,7 @@ impl Default for Identifier {
Self {
kind: IdKind::default(),
length: 4,
- value: 1u32.to_le_bytes().to_vec(),
+ value: 0u32.to_le_bytes().to_vec(),
}
}
}
@@ -367,11 +367,6 @@ mod tests {
assert!(Identifier::numeric(1).is_ok());
}
- #[test]
- fn identifier_with_a_value_of_zero_should_be_invalid() {
- assert!(Identifier::numeric(0).is_err());
- }
-
#[test]
fn identifier_with_a_value_of_non_empty_string_should_be_valid() {
assert!(Identifier::named("test").is_ok());
diff --git a/core/integration/src/test_server.rs
b/core/integration/src/test_server.rs
index 2678eaee..a16b5ce4 100644
--- a/core/integration/src/test_server.rs
+++ b/core/integration/src/test_server.rs
@@ -16,6 +16,15 @@
* under the License.
*/
+use assert_cmd::prelude::CommandCargoExt;
+use async_trait::async_trait;
+use derive_more::Display;
+use futures::executor::block_on;
+use iggy::prelude::UserStatus::Active;
+use iggy::prelude::*;
+use iggy_common::TransportProtocol;
+use rand::Rng;
+use server::configs::config_provider::{ConfigProvider, FileConfigProvider};
use std::collections::HashMap;
use std::fs;
use std::fs::{File, OpenOptions};
@@ -23,20 +32,10 @@ use std::io::Write;
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
-use std::thread::{panicking, sleep};
+use std::thread::{available_parallelism, panicking, sleep};
use std::time::Duration;
-
-use assert_cmd::prelude::CommandCargoExt;
-use async_trait::async_trait;
-use derive_more::Display;
-use futures::executor::block_on;
-use iggy_common::TransportProtocol;
use uuid::Uuid;
-use iggy::prelude::UserStatus::Active;
-use iggy::prelude::*;
-use server::configs::config_provider::{ConfigProvider, FileConfigProvider};
-
pub const SYSTEM_PATH_ENV_VAR: &str = "IGGY_SYSTEM_PATH";
pub const TEST_VERBOSITY_ENV_VAR: &str = "IGGY_TEST_VERBOSE";
pub const IPV6_ENV_VAR: &str = "IGGY_TCP_IPV6";
@@ -99,6 +98,28 @@ impl TestServer {
}
}
+ // Randomly select 4 CPU cores to reduce interference between parallel
tests
+ let cpu_allocation = match available_parallelism() {
+ Ok(parallelism) => {
+ let available_cpus = parallelism.get();
+ if available_cpus >= 4 {
+ let mut rng = rand::thread_rng();
+ let max_start = available_cpus - 4;
+ let start = rng.gen_range(0..=max_start);
+ let end = start + 4;
+ format!("{}..{}", start, end)
+ } else {
+ "all".to_string()
+ }
+ }
+ Err(_) => "0..4".to_string(),
+ };
+
+ envs.insert(
+ "IGGY_SYSTEM_SHARDING_CPU_ALLOCATION".to_string(),
+ cpu_allocation,
+ );
+
if ip_kind == IpAddrKind::V6 {
envs.insert(IPV6_ENV_VAR.to_string(), "true".to_string());
}
@@ -330,6 +351,35 @@ impl TestServer {
}
match file_config_provider.load_config().await {
Ok(config) => {
+ // Verify config contains fresh addresses, not stale
defaults
+ // Default ports: TCP=8090, HTTP=3000, QUIC=8080
+ let tcp_port: u16 = config
+ .tcp
+ .address
+ .split(':')
+ .nth(1)
+ .and_then(|s| s.parse().ok())
+ .unwrap_or(0);
+ let http_port: u16 = config
+ .http
+ .address
+ .split(':')
+ .nth(1)
+ .and_then(|s| s.parse().ok())
+ .unwrap_or(0);
+ let quic_port: u16 = config
+ .quic
+ .address
+ .split(':')
+ .nth(1)
+ .and_then(|s| s.parse().ok())
+ .unwrap_or(0);
+
+ if tcp_port == 8090 || http_port == 3000 || quic_port
== 8080 {
+ sleep(Duration::from_millis(SLEEP_INTERVAL_MS));
+ continue;
+ }
+
loaded_config = Some(config);
break;
}
@@ -340,17 +390,26 @@ impl TestServer {
});
if let Some(config) = config {
- self.server_addrs.push(ServerProtocolAddr::QuicUdp(
- config.quic.address.parse().unwrap(),
- ));
+ let quic_addr: SocketAddr = config.quic.address.parse().unwrap();
+ if quic_addr.port() == 0 {
+ panic!("Quic address port is 0!");
+ }
- self.server_addrs.push(ServerProtocolAddr::RawTcp(
- config.tcp.address.parse().unwrap(),
- ));
+ let tcp_addr: SocketAddr = config.tcp.address.parse().unwrap();
+ if tcp_addr.port() == 0 {
+ panic!("Tcp address port is 0!");
+ }
+
+ let http_addr: SocketAddr = config.http.address.parse().unwrap();
+ if http_addr.port() == 0 {
+ panic!("Http address port is 0!");
+ }
- self.server_addrs.push(ServerProtocolAddr::HttpTcp(
- config.http.address.parse().unwrap(),
- ));
+ self.server_addrs
+ .push(ServerProtocolAddr::QuicUdp(quic_addr));
+ self.server_addrs.push(ServerProtocolAddr::RawTcp(tcp_addr));
+ self.server_addrs
+ .push(ServerProtocolAddr::HttpTcp(http_addr));
} else {
panic!(
"Failed to load config from file {config_path} in
{MAX_PORT_WAIT_DURATION_S} s!"
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 b3e716b0..4e02d614 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
@@ -88,6 +88,8 @@ impl IggyCmdTestCase for TestConsumerGroupCreateCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
+ let stream_details = stream.unwrap();
+ self.stream_id = stream_details.id;
let topic = client
.create_topic(
@@ -101,6 +103,8 @@ impl IggyCmdTestCase for TestConsumerGroupCreateCmd {
)
.await;
assert!(topic.is_ok());
+ let topic_details = topic.unwrap();
+ self.topic_id = topic_details.id;
}
fn get_command(&self) -> IggyCmdCommand {
@@ -186,7 +190,7 @@ pub async fn should_be_successful() {
String::from("main"),
1,
String::from("sync"),
- Some(1),
+ None,
String::from("group1"),
TestStreamId::Numeric,
TestTopicId::Numeric,
@@ -198,7 +202,7 @@ pub async fn should_be_successful() {
String::from("stream"),
3,
String::from("topic"),
- Some(3),
+ None,
String::from("group3"),
TestStreamId::Named,
TestTopicId::Numeric,
@@ -210,7 +214,7 @@ pub async fn should_be_successful() {
String::from("development"),
1,
String::from("probe"),
- Some(7),
+ None,
String::from("group7"),
TestStreamId::Numeric,
TestTopicId::Named,
@@ -222,7 +226,7 @@ pub async fn should_be_successful() {
String::from("production"),
5,
String::from("test"),
- Some(4),
+ None,
String::from("group4"),
TestStreamId::Named,
TestTopicId::Named,
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 1a419177..9c58038e 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
@@ -91,6 +91,8 @@ impl IggyCmdTestCase for TestConsumerGroupDeleteCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
+ let stream_details = stream.unwrap();
+ self.stream_id = stream_details.id;
let topic = client
.create_topic(
@@ -104,6 +106,8 @@ impl IggyCmdTestCase for TestConsumerGroupDeleteCmd {
)
.await;
assert!(topic.is_ok());
+ let topic_details = topic.unwrap();
+ self.topic_id = topic_details.id;
let consumer_group = client
.create_consumer_group(
@@ -113,6 +117,8 @@ impl IggyCmdTestCase for TestConsumerGroupDeleteCmd {
)
.await;
assert!(consumer_group.is_ok());
+ let consumer_group_details = consumer_group.unwrap();
+ self.group_id = consumer_group_details.id;
}
fn get_command(&self) -> IggyCmdCommand {
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 c41231aa..8db78e70 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
@@ -91,12 +91,14 @@ impl IggyCmdTestCase for TestConsumerGroupGetCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
+ let stream_details = stream.unwrap();
+ self.stream_id = stream_details.id;
let topic = client
.create_topic(
&self.stream_id.try_into().unwrap(),
&self.topic_name,
- 0,
+ 1,
Default::default(),
None,
IggyExpiry::NeverExpire,
@@ -104,6 +106,8 @@ impl IggyCmdTestCase for TestConsumerGroupGetCmd {
)
.await;
assert!(topic.is_ok());
+ let topic_details = topic.unwrap();
+ self.topic_id = topic_details.id;
let consumer_group = client
.create_consumer_group(
@@ -113,6 +117,8 @@ impl IggyCmdTestCase for TestConsumerGroupGetCmd {
)
.await;
assert!(consumer_group.is_ok());
+ let consumer_group_details = consumer_group.unwrap();
+ self.group_id = consumer_group_details.id;
}
fn get_command(&self) -> IggyCmdCommand {
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 f7b9dd3a..537787a6 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
@@ -222,7 +222,7 @@ pub async fn should_be_successful() {
String::from("consumer"),
String::from("stream"),
String::from("topic"),
- 1,
+ 0,
using_consumer_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 28328938..4f28bae4 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
@@ -99,6 +99,8 @@ impl IggyCmdTestCase for TestConsumerOffsetSetCmd {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
+ let stream_details = stream.unwrap();
+ self.stream_id = stream_details.id;
let topic = client
.create_topic(
@@ -112,6 +114,8 @@ impl IggyCmdTestCase for TestConsumerOffsetSetCmd {
)
.await;
assert!(topic.is_ok());
+ let topic_details = topic.unwrap();
+ self.topic_id = topic_details.id;
let mut messages = (1..=self.stored_offset + 1)
.filter_map(|id| IggyMessage::from_str(format!("Test message
{id}").as_str()).ok())
@@ -271,7 +275,7 @@ pub async fn should_be_successful() {
String::from("stream"),
3,
String::from("topic"),
- 1,
+ 0,
using_consumer_id,
using_stream_id,
using_topic_id,
diff --git a/core/integration/tests/cli/context/test_context_applied.rs
b/core/integration/tests/cli/context/test_context_applied.rs
index 4e96a69c..0bbbc1dd 100644
--- a/core/integration/tests/cli/context/test_context_applied.rs
+++ b/core/integration/tests/cli/context/test_context_applied.rs
@@ -16,8 +16,8 @@
* under the License.
*/
-use std::collections::HashMap;
-
+use super::common::TestIggyContext;
+use crate::cli::common::{IggyCmdCommand, IggyCmdTest, IggyCmdTestCase};
use assert_cmd::assert::Assert;
use async_trait::async_trait;
use iggy::prelude::ArgsOptional;
@@ -26,10 +26,7 @@ use
iggy_binary_protocol::cli::binary_context::common::ContextConfig;
use integration::test_server::TestServer;
use predicates::str::{contains, starts_with};
use serial_test::parallel;
-
-use crate::cli::common::{IggyCmdCommand, IggyCmdTest, IggyCmdTestCase};
-
-use super::common::TestIggyContext;
+use std::collections::HashMap;
struct TestContextApplied {
set_transport_context: Option<String>,
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 3e2b2486..3531bd7c 100644
--- a/core/integration/tests/cli/message/test_message_flush_command.rs
+++ b/core/integration/tests/cli/message/test_message_flush_command.rs
@@ -169,6 +169,7 @@ impl IggyCmdTestCase for TestMessageFetchCmd {
#[tokio::test]
#[parallel]
+#[ignore = "flush_unsaved_buffer not yet implemented (todo!)"]
pub async fn should_be_successful() {
let mut iggy_cmd_test = IggyCmdTest::default();
@@ -177,7 +178,7 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
for fsync in test_parameters {
iggy_cmd_test
- .execute_test(TestMessageFetchCmd::new("stream", "topic", 1, 1,
fsync))
+ .execute_test(TestMessageFetchCmd::new("stream", "topic", 1, 0,
fsync))
.await;
}
}
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 20f5bcf1..2a4c2b4f 100644
--- a/core/integration/tests/cli/message/test_message_poll_command.rs
+++ b/core/integration/tests/cli/message/test_message_poll_command.rs
@@ -56,8 +56,7 @@ impl TestMessagePollCmd {
show_headers: bool,
headers: (HeaderKey, HeaderValue),
) -> Self {
- assert!(partition_id <= partitions_count);
- assert!(partition_id > 0);
+ assert!(partition_id < partitions_count);
assert!(message_count < messages.len());
Self {
stream_name,
@@ -122,10 +121,9 @@ impl IggyCmdTestCase for TestMessagePollCmd {
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(
- &stream_id,
+ &stream.id.try_into().unwrap(),
&self.topic_name,
self.partitions_count,
Default::default(),
@@ -138,7 +136,6 @@ impl IggyCmdTestCase for TestMessagePollCmd {
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()
@@ -154,8 +151,8 @@ impl IggyCmdTestCase for TestMessagePollCmd {
let send_status = client
.send_messages(
- &stream_id,
- &topic_id,
+ &stream.id.try_into().unwrap(),
+ &topic.id.try_into().unwrap(),
&Partitioning::partition_id(self.partition_id),
&mut messages,
)
@@ -236,14 +233,18 @@ impl IggyCmdTestCase for TestMessagePollCmd {
}
async fn verify_server_state(&self, client: &dyn Client) {
- 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(&stream_id).await;
- assert!(stream.is_ok());
+ if let (Some(stream_id), Some(topic_id)) = (self.actual_stream_id,
self.actual_topic_id) {
+ let topic = client
+ .delete_topic(
+ &stream_id.try_into().unwrap(),
+ &topic_id.try_into().unwrap(),
+ )
+ .await;
+ assert!(topic.is_ok());
+
+ let stream =
client.delete_stream(&stream_id.try_into().unwrap()).await;
+ assert!(stream.is_ok());
+ }
}
}
@@ -271,12 +272,12 @@ pub async fn should_be_successful() {
);
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),
+ (0, 1, PollingStrategy::offset(0), true),
+ (1, 5, PollingStrategy::offset(0), true),
+ (2, 3, PollingStrategy::offset(3), true),
+ (3, 5, PollingStrategy::first(), true),
+ (0, 4, PollingStrategy::last(), true),
+ (1, 3, PollingStrategy::next(), false),
];
iggy_cmd_test.setup().await;
@@ -367,7 +368,7 @@ Options:
{CLAP_INDENT}
Consumer ID can be specified as a consumer name or ID
{CLAP_INDENT}
- [default: 1]
+ [default: 0]
-s, --show-headers
Include the message headers in the output
@@ -418,7 +419,7 @@ Options:
-f, --first Polling strategy - start polling from
the first message in the partition
-l, --last Polling strategy - start polling from
the last message in the partition
-n, --next Polling strategy - start polling from
the next message
- -c, --consumer <CONSUMER> Regular consumer which will poll
messages [default: 1]
+ -c, --consumer <CONSUMER> Regular consumer which will poll
messages [default: 0]
-s, --show-headers Include the message headers in the
output
--output-file <OUTPUT_FILE> Store polled message into file in
binary format
-h, --help Print help (see more with '--help')
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 ace0e485..3f9d029e 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
@@ -36,6 +36,9 @@ pub(super) struct TestMessagePollToFileCmd<'a> {
headers: HashMap<HeaderKey, HeaderValue>,
output_file: String,
cleanup: bool,
+ // These will be populated after creating the resources
+ actual_stream_id: Option<u32>,
+ actual_topic_id: Option<u32>,
}
impl<'a> TestMessagePollToFileCmd<'a> {
@@ -60,6 +63,8 @@ impl<'a> TestMessagePollToFileCmd<'a> {
headers,
output_file: output_file.into(),
cleanup,
+ actual_stream_id: None,
+ actual_topic_id: None,
}
}
@@ -81,11 +86,21 @@ impl<'a> TestMessagePollToFileCmd<'a> {
command.extend(vec!["--output-file".into(), self.output_file.clone()]);
- command.extend(vec![
- self.stream_name.clone(),
- self.topic_name.clone(),
- "1".into(),
- ]);
+ // 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());
+ }
+
+ // 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("0".into());
command
}
@@ -96,14 +111,12 @@ impl IggyCmdTestCase for TestMessagePollToFileCmd<'_> {
async fn prepare_server_state(&mut self, client: &dyn Client) {
let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
-
- let stream_id = Identifier::from_str(self.stream_name.as_str());
- assert!(stream_id.is_ok());
- let stream_id = stream_id.unwrap();
+ let stream = stream.unwrap();
+ self.actual_stream_id = Some(stream.id);
let topic = client
.create_topic(
- &stream_id,
+ &stream.id.try_into().unwrap(),
&self.topic_name,
1,
Default::default(),
@@ -113,10 +126,8 @@ impl IggyCmdTestCase for TestMessagePollToFileCmd<'_> {
)
.await;
assert!(topic.is_ok());
-
- let topic_id = Identifier::from_str(self.topic_name.as_str());
- assert!(topic_id.is_ok());
- let topic_id = topic_id.unwrap();
+ let topic = topic.unwrap();
+ self.actual_topic_id = Some(topic.id);
let mut messages = self
.messages
@@ -133,9 +144,9 @@ impl IggyCmdTestCase for TestMessagePollToFileCmd<'_> {
let send_status = client
.send_messages(
- &stream_id,
- &topic_id,
- &Partitioning::partition_id(1),
+ &stream.id.try_into().unwrap(),
+ &topic.id.try_into().unwrap(),
+ &Partitioning::partition_id(0),
&mut messages,
)
.await;
@@ -156,9 +167,21 @@ impl IggyCmdTestCase for TestMessagePollToFileCmd<'_> {
_ => format!("Polled {} messages", self.message_count),
};
+ let stream_id = if let Some(stream_id) = self.actual_stream_id {
+ format!("{}", stream_id)
+ } else {
+ self.stream_name.clone()
+ };
+
+ let topic_id = if let Some(topic_id) = self.actual_topic_id {
+ format!("{}", topic_id)
+ } else {
+ self.topic_name.clone()
+ };
+
let message_prefix = format!(
- "Executing poll messages from topic ID: {} and stream with ID:
{}\nPolled messages from topic with ID: {} and stream with ID: {} (from
partition with ID: 1)\n{polled_status}",
- self.topic_name, self.stream_name, self.topic_name,
self.stream_name
+ "Executing poll messages from topic ID: {} and stream with ID:
{}\nPolled messages from topic with ID: {} and stream with ID: {} (from
partition with ID: 0)\n{polled_status}",
+ topic_id, stream_id, topic_id, stream_id
);
let message_file = format!("Storing messages to {} binary file",
self.output_file);
let message_count = format!(
@@ -178,19 +201,18 @@ impl IggyCmdTestCase for TestMessagePollToFileCmd<'_> {
}
async fn verify_server_state(&self, client: &dyn Client) {
- let stream_id = Identifier::from_str(self.stream_name.as_str());
- assert!(stream_id.is_ok());
- let stream_id = stream_id.unwrap();
-
- let topic_id = Identifier::from_str(self.topic_name.as_str());
- assert!(topic_id.is_ok());
- let topic_id = topic_id.unwrap();
+ if let (Some(stream_id), Some(topic_id)) = (self.actual_stream_id,
self.actual_topic_id) {
+ let topic = client
+ .delete_topic(
+ &stream_id.try_into().unwrap(),
+ &topic_id.try_into().unwrap(),
+ )
+ .await;
+ assert!(topic.is_ok());
- 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());
+ let stream =
client.delete_stream(&stream_id.try_into().unwrap()).await;
+ assert!(stream.is_ok());
+ }
assert!(Path::new(&self.output_file).is_file());
if self.cleanup {
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 210daa32..27227cd9 100644
--- a/core/integration/tests/cli/message/test_message_send_command.rs
+++ b/core/integration/tests/cli/message/test_message_send_command.rs
@@ -127,12 +127,7 @@ impl TestMessageSendCmd {
fn calculate_partition_id_from_messages_key(&self, messages_key: &[u8]) ->
u32 {
let messages_key_hash = XxHash32::oneshot(0, messages_key);
- let mut partition_id = messages_key_hash % self.partitions_count;
- if partition_id == 0 {
- partition_id = self.partitions_count;
- }
-
- partition_id
+ messages_key_hash % self.partitions_count
}
fn get_partition_id(&self) -> u32 {
@@ -218,32 +213,77 @@ impl IggyCmdTestCase for TestMessageSendCmd {
let topic_details = topic.unwrap().expect("Failed to get topic");
assert_eq!(topic_details.messages_count, self.messages.len() as u64);
- let polled_messages = client
- .poll_messages(
- &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),
- self.messages.len() as u32,
- false,
- )
- .await;
+ // For Balanced partitioning, messages are distributed across all
partitions
+ // so we need to poll from all partitions and collect messages
+ let all_messages = match &self.partitioning {
+ PartitionSelection::Balanced => {
+ let mut collected_messages = Vec::new();
+ for partition_id in 0..self.partitions_count {
+ let polled = client
+ .poll_messages(
+
&self.actual_stream_id.unwrap().try_into().unwrap(),
+ &self.actual_topic_id.unwrap().try_into().unwrap(),
+ Some(partition_id),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ self.messages.len() as u32,
+ false,
+ )
+ .await;
+ if let Ok(polled) = polled {
+ collected_messages.extend(polled.messages);
+ }
+ }
+ collected_messages
+ }
+ _ => {
+ // For specific partition or key-based partitioning
+ let polled_messages = client
+ .poll_messages(
+ &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),
+ self.messages.len() as u32,
+ false,
+ )
+ .await;
+
+ assert!(polled_messages.is_ok());
+ polled_messages.unwrap().messages
+ }
+ };
- assert!(polled_messages.is_ok());
- let polled_messages = polled_messages.unwrap();
- assert_eq!(polled_messages.messages.len(), self.messages.len());
- assert_eq!(
- polled_messages
- .messages
- .iter()
- .map(|m| from_utf8(&m.payload.clone()).unwrap().to_string())
- .collect::<Vec<_>>(),
- self.messages
- );
+ assert_eq!(all_messages.len(), self.messages.len());
+
+ // For Balanced partitioning, messages may arrive in different order
+ // so we just check that all expected messages are present
+ let expected_messages: Vec<String> = self.messages.clone();
+ let received_messages: Vec<String> = all_messages
+ .iter()
+ .map(|m| from_utf8(&m.payload.clone()).unwrap().to_string())
+ .collect();
+
+ match &self.partitioning {
+ PartitionSelection::Balanced => {
+ // For balanced, just check all messages are present (order
may vary)
+ for expected in &expected_messages {
+ assert!(
+ received_messages.contains(expected),
+ "Expected message '{}' not found in received messages",
+ expected
+ );
+ }
+ }
+ _ => {
+ // For specific partition, order should be preserved
+ assert_eq!(received_messages, expected_messages);
+ }
+ }
if let Some(expected_header) = &self.header {
- polled_messages.messages.iter().for_each(|m| {
+ all_messages.iter().for_each(|m| {
assert!(m.user_headers.is_some());
assert_eq!(expected_header,
&m.user_headers_map().unwrap().unwrap());
})
@@ -272,8 +312,8 @@ pub async fn should_be_successful() {
let test_parameters = vec![
(ProvideMessages::AsArgs, PartitionSelection::Balanced),
(ProvideMessages::ViaStdin, PartitionSelection::Balanced),
- (ProvideMessages::ViaStdin, PartitionSelection::Id(1)),
- (ProvideMessages::AsArgs, PartitionSelection::Id(2)),
+ (ProvideMessages::ViaStdin, PartitionSelection::Id(0)),
+ (ProvideMessages::AsArgs, PartitionSelection::Id(1)),
(
ProvideMessages::AsArgs,
PartitionSelection::Key(String::from("some-complex-key")),
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 3327a686..1089bd7f 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
@@ -35,6 +35,9 @@ pub(super) struct TestMessageSendFromFileCmd<'a> {
messages: Vec<&'a str>,
message_count: usize,
headers: HashMap<HeaderKey, HeaderValue>,
+ // These will be populated after creating the resources
+ actual_stream_id: Option<u32>,
+ actual_topic_id: Option<u32>,
}
impl<'a> TestMessageSendFromFileCmd<'a> {
@@ -57,19 +60,33 @@ impl<'a> TestMessageSendFromFileCmd<'a> {
messages: messages.to_owned(),
message_count,
headers,
+ actual_stream_id: None,
+ actual_topic_id: None,
}
}
fn to_args(&self) -> Vec<String> {
- let command = vec![
+ let mut command = vec![
"--input-file".into(),
self.input_file.clone(),
"--partition-id".into(),
- "1".into(),
- self.stream_name.clone(),
- self.topic_name.clone(),
+ "0".into(),
];
+ // 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());
+ }
+
+ // 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
}
}
@@ -77,16 +94,17 @@ 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).await;
- assert!(stream.is_ok());
-
- let stream_id = Identifier::from_str(self.stream_name.as_str());
- 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,
1,
Default::default(),
@@ -94,11 +112,9 @@ impl IggyCmdTestCase for TestMessageSendFromFileCmd<'_> {
IggyExpiry::NeverExpire,
MaxTopicSize::ServerDefault,
)
- .await;
- assert!(topic.is_ok());
-
- let topic_id = Identifier::from_str(self.topic_name.as_str());
- assert!(topic_id.is_ok());
+ .await
+ .expect("Failed to create topic");
+ self.actual_topic_id = Some(topic.id);
if self.initialize {
let file = tokio::fs::OpenOptions::new()
@@ -153,9 +169,21 @@ impl IggyCmdTestCase for TestMessageSendFromFileCmd<'_> {
}
fn verify_command(&self, command_state: Assert) {
+ let stream_id = if let Some(stream_id) = self.actual_stream_id {
+ format!("{}", stream_id)
+ } else {
+ self.stream_name.clone()
+ };
+
+ let topic_id = if let Some(topic_id) = self.actual_topic_id {
+ format!("{}", topic_id)
+ } else {
+ self.topic_name.clone()
+ };
+
let message_prefix = format!(
"Executing send messages to topic with ID: {} and stream with ID:
{}\n",
- self.topic_name, self.stream_name
+ topic_id, stream_id
);
let message_read = format!("Read [0-9]+ bytes from {} file",
self.input_file);
let message_created = format!(
@@ -167,7 +195,7 @@ impl IggyCmdTestCase for TestMessageSendFromFileCmd<'_> {
);
let message_sent = format!(
"Sent messages to topic with ID: {} and stream with ID: {}\n",
- self.topic_name, self.stream_name
+ topic_id, stream_id
);
command_state
@@ -179,50 +207,49 @@ impl IggyCmdTestCase for TestMessageSendFromFileCmd<'_> {
}
async fn verify_server_state(&self, client: &dyn Client) {
- let stream_id = Identifier::from_str(self.stream_name.as_str());
- assert!(stream_id.is_ok());
- let stream_id = stream_id.unwrap();
+ if let (Some(stream_id), Some(topic_id)) = (self.actual_stream_id,
self.actual_topic_id) {
+ let messages = client
+ .poll_messages(
+ &stream_id.try_into().unwrap(),
+ &topic_id.try_into().unwrap(),
+ Some(0),
+ &Consumer::new(Identifier::default()),
+ &PollingStrategy::offset(0),
+ self.message_count as u32 * 2,
+ true,
+ )
+ .await;
+ assert!(messages.is_ok());
+ let messages = messages.unwrap();
- let topic_id = Identifier::from_str(self.topic_name.as_str());
- assert!(topic_id.is_ok());
- let topic_id = topic_id.unwrap();
+ // Check if there are only the expected number of messages
+ assert_eq!(messages.messages.len(), self.message_count);
- let messages = client
- .poll_messages(
- &stream_id,
- &topic_id,
- Some(1),
- &Consumer::new(Identifier::default()),
- &PollingStrategy::offset(0),
- self.message_count as u32 * 2,
- true,
- )
- .await;
- assert!(messages.is_ok());
- let messages = messages.unwrap();
+ // Check message order and content (payload and headers)
+ for (i, message) in messages.messages.iter().enumerate() {
+ assert_eq!(
+ message.payload,
+ Bytes::from(self.messages[i].as_bytes().to_vec())
+ );
+ assert_eq!(
+ message.user_headers_map().unwrap().is_some(),
+ !self.headers.is_empty()
+ );
+ assert_eq!(&message.user_headers_map().unwrap().unwrap(),
&self.headers);
+ }
- // Check if there are only the expected number of messages
- assert_eq!(messages.messages.len(), self.message_count);
+ let topic_delete = client
+ .delete_topic(
+ &stream_id.try_into().unwrap(),
+ &topic_id.try_into().unwrap(),
+ )
+ .await;
+ assert!(topic_delete.is_ok());
- // Check message order and content (payload and headers)
- for (i, message) in messages.messages.iter().enumerate() {
- assert_eq!(
- message.payload,
- Bytes::from(self.messages[i].as_bytes().to_vec())
- );
- assert_eq!(
- message.user_headers_map().unwrap().is_some(),
- !self.headers.is_empty()
- );
- assert_eq!(&message.user_headers_map().unwrap().unwrap(),
&self.headers);
+ let stream_delete =
client.delete_stream(&stream_id.try_into().unwrap()).await;
+ assert!(stream_delete.is_ok());
}
- let topic_delete = client.delete_topic(&stream_id, &topic_id).await;
- assert!(topic_delete.is_ok());
-
- let stream_delete = client.delete_stream(&stream_id).await;
- assert!(stream_delete.is_ok());
-
let file_removal = std::fs::remove_file(&self.input_file);
assert!(file_removal.is_ok());
}
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 1fdfc3c3..8e726cd8 100644
--- a/core/integration/tests/cli/partition/test_partition_delete_command.rs
+++ b/core/integration/tests/cli/partition/test_partition_delete_command.rs
@@ -57,19 +57,11 @@ impl TestPartitionDeleteCmd {
fn to_args(&self) -> Vec<String> {
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());
- }
+ // Always use stream name for consistency with other tests
+ command.push(self.stream_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());
- }
+ // Always use topic name for consistency with other tests
+ command.push(self.topic_name.clone());
command.push(format!("{}", self.new_partitions));
@@ -108,18 +100,6 @@ impl IggyCmdTestCase for TestPartitionDeleteCmd {
}
fn verify_command(&self, command_state: Assert) {
- let stream_id = if let Some(stream_id) = self.actual_stream_id {
- format!("{}", stream_id)
- } else {
- self.stream_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");
if self.new_partitions > 1 {
partitions.push('s');
@@ -127,7 +107,12 @@ impl IggyCmdTestCase for TestPartitionDeleteCmd {
let message = format!(
"Executing delete {} {partitions} for topic with ID: {} and stream
with ID: {}\nDeleted {} {partitions} for topic with ID: {} and stream with ID:
{}\n",
- self.new_partitions, topic_id, stream_id, self.new_partitions,
topic_id, stream_id
+ self.new_partitions,
+ self.topic_name,
+ self.stream_name,
+ self.new_partitions,
+ self.topic_name,
+ self.stream_name
);
command_state.success().stdout(diff(message));
@@ -144,14 +129,10 @@ impl IggyCmdTestCase for TestPartitionDeleteCmd {
let topic_details = topic.unwrap().expect("Failed to get topic");
assert_eq!(topic_details.name, self.topic_name);
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 {
- assert_eq!(
- topic_details.partitions_count,
- self.partitions_count - self.new_partitions
- );
- }
+ assert_eq!(
+ topic_details.partitions_count,
+ self.partitions_count - self.new_partitions
+ );
assert_eq!(topic_details.messages_count, 0);
let topic = client
@@ -203,8 +184,8 @@ pub async fn should_be_successful() {
.execute_test(TestPartitionDeleteCmd::new(
String::from("production"),
String::from("test"),
+ 5,
3,
- 7,
))
.await;
}
diff --git a/core/integration/tests/cli/stream/test_stream_create_command.rs
b/core/integration/tests/cli/stream/test_stream_create_command.rs
index 70e18f02..b657863b 100644
--- a/core/integration/tests/cli/stream/test_stream_create_command.rs
+++ b/core/integration/tests/cli/stream/test_stream_create_command.rs
@@ -25,12 +25,15 @@ use serial_test::parallel;
struct TestStreamCreateCmd {
stream_id: Option<u32>,
- name: String,
+ stream_name: String,
}
impl TestStreamCreateCmd {
fn new(stream_id: Option<u32>, name: String) -> Self {
- Self { stream_id, name }
+ Self {
+ stream_id,
+ stream_name: name,
+ }
}
fn to_args(&self) -> Vec<String> {
@@ -41,7 +44,7 @@ impl TestStreamCreateCmd {
args.push(format!("{stream_id}"));
}
- args.push(self.name.clone());
+ args.push(self.stream_name.clone());
args
}
@@ -60,14 +63,11 @@ impl IggyCmdTestCase for TestStreamCreateCmd {
}
fn verify_command(&self, command_state: Assert) {
- let stream_id = match self.stream_id {
- Some(stream_id) => format!("ID: {stream_id}"),
- None => "ID auto incremented".to_string(),
- };
+ let stream_id = "ID auto incremented";
let message = format!(
"Executing create stream with name: {} and {}\nStream with name:
{} and {} created\n",
- self.name, stream_id, self.name, stream_id
+ self.stream_name, stream_id, self.stream_name, stream_id
);
command_state.success().stdout(diff(message));
@@ -75,17 +75,14 @@ impl IggyCmdTestCase for TestStreamCreateCmd {
async fn verify_server_state(&self, client: &dyn Client) {
let stream = client
- .get_stream(&self.name.clone().try_into().unwrap())
+ .get_stream(&self.stream_name.clone().try_into().unwrap())
.await;
assert!(stream.is_ok());
let stream = stream.unwrap().expect("Stream not found");
- assert_eq!(stream.name, self.name);
- if let Some(stream_id) = self.stream_id {
- assert_eq!(stream.id, stream_id);
- }
+ assert_eq!(stream.name, self.stream_name);
let delete = client
- .delete_stream(&self.name.clone().try_into().unwrap())
+ .delete_stream(&self.stream_name.clone().try_into().unwrap())
.await;
assert!(delete.is_ok());
}
@@ -98,7 +95,7 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
- .execute_test(TestStreamCreateCmd::new(Some(123),
String::from("main")))
+ .execute_test(TestStreamCreateCmd::new(None, String::from("main")))
.await;
iggy_cmd_test.setup().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 c1f3ebc4..aa548aeb 100644
--- a/core/integration/tests/cli/stream/test_stream_delete_command.rs
+++ b/core/integration/tests/cli/stream/test_stream_delete_command.rs
@@ -28,7 +28,7 @@ use serial_test::parallel;
struct TestStreamDeleteCmd {
stream_id: u32,
- name: String,
+ stream_name: String,
using_identifier: TestStreamId,
}
@@ -36,14 +36,14 @@ impl TestStreamDeleteCmd {
fn new(stream_id: u32, name: String, using_identifier: TestStreamId) ->
Self {
Self {
stream_id,
- name,
+ stream_name: name,
using_identifier,
}
}
fn to_arg(&self) -> String {
match self.using_identifier {
- TestStreamId::Named => self.name.clone(),
+ TestStreamId::Named => self.stream_name.clone(),
TestStreamId::Numeric => format!("{}", self.stream_id),
}
}
@@ -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).await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
}
@@ -68,7 +68,7 @@ impl IggyCmdTestCase for TestStreamDeleteCmd {
let message = match self.using_identifier {
TestStreamId::Named => format!(
"Executing delete stream with ID: {}\nStream with ID: {}
deleted\n",
- self.name, self.name
+ self.stream_name, self.stream_name
),
TestStreamId::Numeric => format!(
"Executing delete stream with ID: {}\nStream with ID: {}
deleted\n",
@@ -95,14 +95,14 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestStreamDeleteCmd::new(
- 1,
+ 0,
String::from("testing"),
- TestStreamId::Numeric,
+ TestStreamId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestStreamDeleteCmd::new(
- 2,
+ 0,
String::from("production"),
TestStreamId::Named,
))
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 ff9f5cc4..a46543ef 100644
--- a/core/integration/tests/cli/stream/test_stream_get_command.rs
+++ b/core/integration/tests/cli/stream/test_stream_get_command.rs
@@ -28,7 +28,7 @@ use serial_test::parallel;
struct TestStreamGetCmd {
stream_id: u32,
- name: String,
+ stream_name: String,
using_identifier: TestStreamId,
}
@@ -36,14 +36,14 @@ impl TestStreamGetCmd {
fn new(stream_id: u32, name: String, using_identifier: TestStreamId) ->
Self {
Self {
stream_id,
- name,
+ stream_name: name,
using_identifier,
}
}
fn to_arg(&self) -> String {
match self.using_identifier {
- TestStreamId::Named => self.name.clone(),
+ TestStreamId::Named => self.stream_name.clone(),
TestStreamId::Numeric => format!("{}", self.stream_id),
}
}
@@ -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).await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
}
@@ -66,7 +66,10 @@ impl IggyCmdTestCase for TestStreamGetCmd {
fn verify_command(&self, command_state: Assert) {
let start_message = match self.using_identifier {
- TestStreamId::Named => format!("Executing get stream with ID:
{}\n", self.name.clone()),
+ TestStreamId::Named => format!(
+ "Executing get stream with ID: {}\n",
+ self.stream_name.clone()
+ ),
TestStreamId::Numeric => format!("Executing get stream with ID:
{}\n", self.stream_id),
};
@@ -74,10 +77,9 @@ impl IggyCmdTestCase for TestStreamGetCmd {
.success()
.stdout(starts_with(start_message))
.stdout(contains(format!(
- "Stream ID | {}",
- self.stream_id
+ "Stream name | {}",
+ self.stream_name
)))
- .stdout(contains(format!("Stream name | {}", self.name)))
.stdout(contains("Stream size | 0"))
.stdout(contains("Stream message count | 0"))
.stdout(contains("Stream topics count | 0"));
@@ -94,16 +96,16 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestStreamGetCmd::new(
- 1,
+ 0,
String::from("production"),
TestStreamId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestStreamGetCmd::new(
- 2,
+ 0,
String::from("testing"),
- TestStreamId::Numeric,
+ TestStreamId::Named,
))
.await;
}
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 d605f8f8..78f897f2 100644
--- a/core/integration/tests/cli/stream/test_stream_purge_command.rs
+++ b/core/integration/tests/cli/stream/test_stream_purge_command.rs
@@ -41,7 +41,7 @@ impl TestStreamPurgeCmd {
stream_id,
stream_name: name,
using_identifier,
- topic_id: 1,
+ topic_id: 0,
topic_name: String::from("test_topic"),
}
}
@@ -62,7 +62,7 @@ impl IggyCmdTestCase for TestStreamPurgeCmd {
let topic = client
.create_topic(
- &self.stream_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
&self.topic_name,
10,
Default::default(),
@@ -80,15 +80,17 @@ impl IggyCmdTestCase for TestStreamPurgeCmd {
let send_status = client
.send_messages(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
+ &self.topic_name.clone().try_into().unwrap(),
&Partitioning::default(),
&mut messages,
)
.await;
assert!(send_status.is_ok());
- let stream_state =
client.get_stream(&self.stream_id.try_into().unwrap()).await;
+ let stream_state = client
+ .get_stream(&self.stream_name.clone().try_into().unwrap())
+ .await;
assert!(stream_state.is_ok());
let stream_state = stream_state.unwrap().expect("Stream not found");
assert!(stream_state.size > 0);
@@ -116,13 +118,15 @@ impl IggyCmdTestCase for TestStreamPurgeCmd {
}
async fn verify_server_state(&self, client: &dyn Client) {
- let stream_state =
client.get_stream(&self.stream_id.try_into().unwrap()).await;
+ let stream_state = client
+ .get_stream(&self.stream_name.clone().try_into().unwrap())
+ .await;
assert!(stream_state.is_ok());
let stream_state = stream_state.unwrap().expect("Stream not found");
assert_eq!(stream_state.size, 0);
let stream_delete = client
- .delete_stream(&self.stream_id.try_into().unwrap())
+ .delete_stream(&self.stream_name.clone().try_into().unwrap())
.await;
assert!(stream_delete.is_ok());
}
@@ -136,16 +140,16 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestStreamPurgeCmd::new(
- 1,
+ 0,
String::from("production"),
TestStreamId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestStreamPurgeCmd::new(
- 2,
+ 1,
String::from("testing"),
- TestStreamId::Numeric,
+ TestStreamId::Named,
))
.await;
}
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 bf5f7d93..94eed72e 100644
--- a/core/integration/tests/cli/stream/test_stream_update_command.rs
+++ b/core/integration/tests/cli/stream/test_stream_update_command.rs
@@ -28,7 +28,7 @@ use serial_test::parallel;
struct TestStreamUpdateCmd {
stream_id: u32,
- name: String,
+ stream_name: String,
new_name: String,
using_identifier: TestStreamId,
}
@@ -37,7 +37,7 @@ impl TestStreamUpdateCmd {
fn new(stream_id: u32, name: String, new_name: String, using_identifier:
TestStreamId) -> Self {
Self {
stream_id,
- name,
+ stream_name: name,
new_name,
using_identifier,
}
@@ -45,7 +45,7 @@ impl TestStreamUpdateCmd {
fn to_args(&self) -> Vec<String> {
match self.using_identifier {
- TestStreamId::Named => vec![self.name.clone(),
self.new_name.clone()],
+ TestStreamId::Named => vec![self.stream_name.clone(),
self.new_name.clone()],
TestStreamId::Numeric => {
vec![format!("{}", self.stream_id), self.new_name.clone()]
}
@@ -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).await;
+ let stream = client.create_stream(&self.stream_name).await;
assert!(stream.is_ok());
}
@@ -72,7 +72,7 @@ impl IggyCmdTestCase for TestStreamUpdateCmd {
let message = match self.using_identifier {
TestStreamId::Named => format!(
"Executing update stream with ID: {} and name: {}\nStream with
ID: {} updated name: {}\n",
- self.name, self.new_name, self.name, self.new_name
+ self.stream_name, self.new_name, self.stream_name,
self.new_name
),
TestStreamId::Numeric => format!(
"Executing update stream with ID: {} and name: {}\nStream with
ID: {} updated name: {}\n",
@@ -84,10 +84,17 @@ impl IggyCmdTestCase for TestStreamUpdateCmd {
}
async fn verify_server_state(&self, client: &dyn Client) {
- let stream =
client.get_stream(&self.stream_id.try_into().unwrap()).await;
+ let stream = client
+ .get_stream(&self.new_name.clone().try_into().unwrap())
+ .await;
assert!(stream.is_ok());
let stream = stream.unwrap().expect("Stream not found");
assert_eq!(stream.name, self.new_name);
+
+ let delete = client
+ .delete_stream(&self.new_name.clone().try_into().unwrap())
+ .await;
+ assert!(delete.is_ok());
}
}
@@ -99,15 +106,15 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestStreamUpdateCmd::new(
- 1,
+ 0,
String::from("testing"),
String::from("development"),
- TestStreamId::Numeric,
+ TestStreamId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestStreamUpdateCmd::new(
- 2,
+ 0,
String::from("production"),
String::from("prototype"),
TestStreamId::Named,
diff --git a/core/integration/tests/cli/system/test_stats_command.rs
b/core/integration/tests/cli/system/test_stats_command.rs
index bc9b6db2..4f921c7d 100644
--- a/core/integration/tests/cli/system/test_stats_command.rs
+++ b/core/integration/tests/cli/system/test_stats_command.rs
@@ -96,7 +96,7 @@ impl IggyCmdTestCase for TestStatsCmd {
.stdout(contains("Partitions Count | 5"))
.stdout(contains("Segments Count | 5"))
.stdout(contains("Message Count | 0"))
- .stdout(contains("Clients Count | 2")) // 2
clients are connected during test
+ // Note: Client count can vary due to connection
lifecycle; at least 2 expected
.stdout(contains("Consumer Groups Count | 0"));
}
TestStatsCmdOutput::Set(GetStatsOutput::List) => {
@@ -107,7 +107,6 @@ impl IggyCmdTestCase for TestStatsCmd {
.stdout(contains("Partitions Count|5"))
.stdout(contains("Segments Count|5"))
.stdout(contains("Message Count|0"))
- .stdout(contains("Clients Count|2")) // 2 clients are
connected during test
.stdout(contains("Consumer Groups Count|0"));
}
TestStatsCmdOutput::Set(GetStatsOutput::Json) => {
@@ -118,7 +117,6 @@ impl IggyCmdTestCase for TestStatsCmd {
.stdout(contains(r#""partitions_count": 5"#))
.stdout(contains(r#""segments_count": 5"#))
.stdout(contains(r#""messages_count": 0"#))
- .stdout(contains(r#""clients_count": 2"#)) // 2 clients
are connected during test
.stdout(contains(r#""consumer_groups_count": 0"#));
}
TestStatsCmdOutput::Set(GetStatsOutput::Toml) => {
@@ -129,7 +127,6 @@ impl IggyCmdTestCase for TestStatsCmd {
.stdout(contains("partitions_count = 5"))
.stdout(contains("segments_count = 5"))
.stdout(contains("messages_count = 0"))
- .stdout(contains("clients_count = 2")) // 2 clients are
connected during test
.stdout(contains("consumer_groups_count = 0"));
}
}
@@ -137,11 +134,11 @@ impl IggyCmdTestCase for TestStatsCmd {
async fn verify_server_state(&self, client: &dyn Client) {
let topic = client
- .delete_topic(&1.try_into().unwrap(), &1.try_into().unwrap())
+ .delete_topic(&0.try_into().unwrap(), &0.try_into().unwrap())
.await;
assert!(topic.is_ok());
- let stream = client.delete_stream(&1.try_into().unwrap()).await;
+ let stream = client.delete_stream(&0.try_into().unwrap()).await;
assert!(stream.is_ok());
}
}
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 6cb045c3..96e100d4 100644
--- a/core/integration/tests/cli/topic/test_topic_create_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_create_command.rs
@@ -77,11 +77,6 @@ impl TestTopicCreateCmd {
fn to_args(&self) -> Vec<String> {
let mut args = Vec::new();
- if let Some(topic_id) = self.topic_id {
- args.push("-t".to_string());
- args.push(format!("{topic_id}"));
- };
-
match self.using_identifier {
TestStreamId::Numeric => args.extend(vec![format!("{}",
self.stream_id)]),
TestStreamId::Named => args.extend(vec![self.stream_name.clone()]),
@@ -119,10 +114,6 @@ impl IggyCmdTestCase for TestTopicCreateCmd {
TestStreamId::Named => self.stream_name.clone(),
};
let partitions_count = self.partitions_count;
- let topic_id = match self.topic_id {
- Some(topic_id) => format!("ID: {topic_id}"),
- None => "ID auto incremented".to_string(),
- };
let topic_name = &self.topic_name;
let compression_algorithm = &self.compression_algorithm;
let message_expiry = (match &self.message_expiry {
@@ -136,9 +127,9 @@ impl IggyCmdTestCase for TestTopicCreateCmd {
let replication_factor = self.replication_factor;
let message = format!(
- "Executing create topic with name: {topic_name}, {topic_id},
message expiry: {message_expiry}, compression algorithm:
{compression_algorithm}, \
+ "Executing 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}\n\
- Topic with name: {topic_name}, {topic_id}, partitions count:
{partitions_count}, compression algorithm: {compression_algorithm}, message
expiry: {message_expiry}, \
+ Topic with name: {topic_name}, partitions count:
{partitions_count}, compression algorithm: {compression_algorithm}, message
expiry: {message_expiry}, \
max topic size: {max_topic_size}, replication factor:
{replication_factor} created in stream with ID: {stream_id}\n",
);
@@ -148,7 +139,7 @@ impl IggyCmdTestCase for TestTopicCreateCmd {
async fn verify_server_state(&self, client: &dyn Client) {
let topic = client
.get_topic(
- &self.stream_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
&self.topic_name.clone().try_into().unwrap(),
)
.await;
@@ -157,9 +148,6 @@ impl IggyCmdTestCase for TestTopicCreateCmd {
assert_eq!(topic_details.name, self.topic_name);
assert_eq!(topic_details.partitions_count, self.partitions_count);
assert_eq!(topic_details.messages_count, 0);
- if let Some(topic_id) = self.topic_id {
- assert_eq!(topic_details.id, topic_id);
- }
if self.message_expiry.is_some() {
let duration: Duration = *self
@@ -177,14 +165,14 @@ impl IggyCmdTestCase for TestTopicCreateCmd {
let delete_topic = client
.delete_topic(
- &self.stream_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
&self.topic_name.clone().try_into().unwrap(),
)
.await;
assert!(delete_topic.is_ok());
let delete_stream = client
- .delete_stream(&self.stream_id.try_into().unwrap())
+ .delete_stream(&self.stream_name.clone().try_into().unwrap())
.await;
assert!(delete_stream.is_ok());
}
@@ -198,7 +186,7 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestTopicCreateCmd::new(
- 1,
+ 0,
String::from("main"),
None,
String::from("sync"),
@@ -207,14 +195,14 @@ pub async fn should_be_successful() {
None,
MaxTopicSize::ServerDefault,
1,
- TestStreamId::Numeric,
+ TestStreamId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestTopicCreateCmd::new(
- 2,
+ 1,
String::from("testing"),
- Some(2),
+ None,
String::from("topic"),
5,
Default::default(),
@@ -226,7 +214,7 @@ pub async fn should_be_successful() {
.await;
iggy_cmd_test
.execute_test(TestTopicCreateCmd::new(
- 3,
+ 2,
String::from("prod"),
None,
String::from("named"),
@@ -240,9 +228,9 @@ pub async fn should_be_successful() {
.await;
iggy_cmd_test
.execute_test(TestTopicCreateCmd::new(
- 4,
+ 3,
String::from("big"),
- Some(1),
+ None,
String::from("probe"),
2,
Default::default(),
@@ -254,7 +242,7 @@ pub async fn should_be_successful() {
]),
MaxTopicSize::Custom(IggyByteSize::from_str("2GiB").unwrap()),
1,
- TestStreamId::Numeric,
+ TestStreamId::Named,
))
.await;
}
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 9f3e5c0e..a332b1b6 100644
--- a/core/integration/tests/cli/topic/test_topic_delete_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_delete_command.rs
@@ -79,7 +79,7 @@ impl IggyCmdTestCase for TestTopicDeleteCmd {
let topic = client
.create_topic(
- &self.stream_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
&self.topic_name,
1,
Default::default(),
@@ -121,7 +121,9 @@ impl IggyCmdTestCase for TestTopicDeleteCmd {
}
async fn verify_server_state(&self, client: &dyn Client) {
- let topic =
client.get_topics(&self.stream_id.try_into().unwrap()).await;
+ let topic = client
+ .get_topics(&self.stream_name.clone().try_into().unwrap())
+ .await;
assert!(topic.is_ok());
let topics = topic.unwrap();
assert!(topics.is_empty());
@@ -136,19 +138,19 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestTopicDeleteCmd::new(
- 1,
+ 0,
String::from("main"),
- 1,
+ 0,
String::from("sync"),
- TestStreamId::Numeric,
- TestTopicId::Numeric,
+ TestStreamId::Named,
+ TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestTopicDeleteCmd::new(
- 2,
+ 1,
String::from("testing"),
- 2,
+ 1,
String::from("topic"),
TestStreamId::Named,
TestTopicId::Named,
@@ -156,21 +158,21 @@ pub async fn should_be_successful() {
.await;
iggy_cmd_test
.execute_test(TestTopicDeleteCmd::new(
- 3,
+ 2,
String::from("prod"),
- 1,
+ 0,
String::from("named"),
TestStreamId::Named,
- TestTopicId::Numeric,
+ TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestTopicDeleteCmd::new(
- 4,
+ 3,
String::from("big"),
- 1,
+ 0,
String::from("probe"),
- TestStreamId::Numeric,
+ TestStreamId::Named,
TestTopicId::Named,
))
.await;
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 0fec1f66..c3185219 100644
--- a/core/integration/tests/cli/topic/test_topic_get_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_get_command.rs
@@ -79,7 +79,7 @@ impl IggyCmdTestCase for TestTopicGetCmd {
let topic = client
.create_topic(
- &self.stream_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
&self.topic_name,
1,
Default::default(),
@@ -116,7 +116,7 @@ impl IggyCmdTestCase for TestTopicGetCmd {
command_state
.success()
.stdout(starts_with(start_message))
- .stdout(contains(format!("Topic id | {}",
self.topic_id)))
+ .stdout(contains("Topic id | 0"))
.stdout(contains(format!(
"Topic name | {}",
self.topic_name
@@ -131,14 +131,14 @@ impl IggyCmdTestCase for TestTopicGetCmd {
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(),
+ &self.stream_name.clone().try_into().unwrap(),
+ &self.topic_name.clone().try_into().unwrap(),
)
.await;
assert!(topic.is_ok());
let stream = client
- .delete_stream(&self.stream_id.try_into().unwrap())
+ .delete_stream(&self.stream_name.clone().try_into().unwrap())
.await;
assert!(stream.is_ok());
}
@@ -152,19 +152,19 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestTopicGetCmd::new(
- 1,
+ 0,
String::from("main"),
- 1,
+ 0,
String::from("sync"),
- TestStreamId::Numeric,
- TestTopicId::Numeric,
+ TestStreamId::Named,
+ TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestTopicGetCmd::new(
- 2,
+ 1,
String::from("customer"),
- 4,
+ 3,
String::from("probe"),
TestStreamId::Named,
TestTopicId::Named,
@@ -172,22 +172,22 @@ pub async fn should_be_successful() {
.await;
iggy_cmd_test
.execute_test(TestTopicGetCmd::new(
- 2,
+ 1,
String::from("development"),
- 3,
+ 2,
String::from("testing"),
- TestStreamId::Numeric,
+ TestStreamId::Named,
TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestTopicGetCmd::new(
- 7,
+ 6,
String::from("other"),
- 2,
+ 1,
String::from("topic"),
TestStreamId::Named,
- TestTopicId::Numeric,
+ TestTopicId::Named,
))
.await;
}
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 b025308e..0f394677 100644
--- a/core/integration/tests/cli/topic/test_topic_list_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_list_command.rs
@@ -76,7 +76,7 @@ impl IggyCmdTestCase for TestTopicListCmd {
let topic = client
.create_topic(
- &self.stream_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
&self.topic_name,
1,
Default::default(),
@@ -114,14 +114,14 @@ impl IggyCmdTestCase for TestTopicListCmd {
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(),
+ &self.stream_name.clone().try_into().unwrap(),
+ &self.topic_name.clone().try_into().unwrap(),
)
.await;
assert!(topic.is_ok());
let stream = client
- .delete_stream(&self.stream_id.try_into().unwrap())
+ .delete_stream(&self.stream_name.clone().try_into().unwrap())
.await;
assert!(stream.is_ok());
}
@@ -135,19 +135,19 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestTopicListCmd::new(
- 1,
+ 0,
String::from("main"),
- 1,
+ 0,
String::from("sync"),
- TestStreamId::Numeric,
+ TestStreamId::Named,
OutputFormat::Default,
))
.await;
iggy_cmd_test
.execute_test(TestTopicListCmd::new(
- 2,
+ 1,
String::from("customer"),
- 3,
+ 0,
String::from("topic"),
TestStreamId::Named,
OutputFormat::List,
@@ -155,11 +155,11 @@ pub async fn should_be_successful() {
.await;
iggy_cmd_test
.execute_test(TestTopicListCmd::new(
- 3,
+ 2,
String::from("production"),
- 1,
+ 0,
String::from("data"),
- TestStreamId::Numeric,
+ TestStreamId::Named,
OutputFormat::Table,
))
.await;
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 ee39cbe6..ddc9e2c4 100644
--- a/core/integration/tests/cli/topic/test_topic_purge_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_purge_command.rs
@@ -78,7 +78,7 @@ impl IggyCmdTestCase for TestTopicPurgeCmd {
let topic = client
.create_topic(
- &self.stream_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
&self.topic_name,
10,
Default::default(),
@@ -96,8 +96,8 @@ impl IggyCmdTestCase for TestTopicPurgeCmd {
let send_status = client
.send_messages(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
+ &self.topic_name.clone().try_into().unwrap(),
&Partitioning::default(),
&mut messages,
)
@@ -106,8 +106,8 @@ impl IggyCmdTestCase for TestTopicPurgeCmd {
let topic_state = client
.get_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
+ &self.topic_name.clone().try_into().unwrap(),
)
.await;
assert!(topic_state.is_ok());
@@ -144,8 +144,8 @@ impl IggyCmdTestCase for TestTopicPurgeCmd {
async fn verify_server_state(&self, client: &dyn Client) {
let topic_state = client
.get_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
+ &self.topic_name.clone().try_into().unwrap(),
)
.await;
assert!(topic_state.is_ok());
@@ -154,14 +154,14 @@ impl IggyCmdTestCase for TestTopicPurgeCmd {
let topic_delete = client
.delete_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
+ &self.topic_name.clone().try_into().unwrap(),
)
.await;
assert!(topic_delete.is_ok());
let stream_delete = client
- .delete_stream(&self.stream_id.try_into().unwrap())
+ .delete_stream(&self.stream_name.clone().try_into().unwrap())
.await;
assert!(stream_delete.is_ok());
}
@@ -175,19 +175,19 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestTopicPurgeCmd::new(
- 1,
+ 0,
String::from("main"),
- 1,
+ 0,
String::from("sync"),
- TestStreamId::Numeric,
- TestTopicId::Numeric,
+ TestStreamId::Named,
+ TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestTopicPurgeCmd::new(
- 2,
+ 1,
String::from("testing"),
- 2,
+ 0,
String::from("topic"),
TestStreamId::Named,
TestTopicId::Named,
@@ -195,21 +195,21 @@ pub async fn should_be_successful() {
.await;
iggy_cmd_test
.execute_test(TestTopicPurgeCmd::new(
- 3,
+ 2,
String::from("prod"),
- 1,
+ 0,
String::from("named"),
TestStreamId::Named,
- TestTopicId::Numeric,
+ TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestTopicPurgeCmd::new(
- 4,
+ 3,
String::from("big"),
- 1,
+ 0,
String::from("probe"),
- TestStreamId::Numeric,
+ TestStreamId::Named,
TestTopicId::Named,
))
.await;
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 f37e957e..2d62a49e 100644
--- a/core/integration/tests/cli/topic/test_topic_update_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_update_command.rs
@@ -137,7 +137,7 @@ impl IggyCmdTestCase for TestTopicUpdateCmd {
let topic = client
.create_topic(
- &self.stream_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
&self.topic_name,
1,
self.compression_algorithm,
@@ -195,14 +195,13 @@ impl IggyCmdTestCase for TestTopicUpdateCmd {
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.stream_name.clone().try_into().unwrap(),
+ &self.topic_new_name.clone().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_new_name);
- assert_eq!(topic_details.id, self.topic_id);
assert_eq!(topic_details.messages_count, 0);
if self.topic_new_message_expiry.is_some() {
@@ -221,14 +220,14 @@ impl IggyCmdTestCase for TestTopicUpdateCmd {
let topic = client
.delete_topic(
- &self.stream_id.try_into().unwrap(),
- &self.topic_id.try_into().unwrap(),
+ &self.stream_name.clone().try_into().unwrap(),
+ &self.topic_new_name.clone().try_into().unwrap(),
)
.await;
assert!(topic.is_ok());
let stream = client
- .delete_stream(&self.stream_id.try_into().unwrap())
+ .delete_stream(&self.stream_name.clone().try_into().unwrap())
.await;
assert!(stream.is_ok());
}
@@ -241,9 +240,9 @@ pub async fn should_be_successful() {
iggy_cmd_test.setup().await;
iggy_cmd_test
.execute_test(TestTopicUpdateCmd::new(
- 1,
+ 0,
String::from("main"),
- 1,
+ 0,
String::from("sync"),
Default::default(),
None,
@@ -254,15 +253,15 @@ pub async fn should_be_successful() {
None,
MaxTopicSize::Custom(IggyByteSize::from_str("2GiB").unwrap()),
1,
- TestStreamId::Numeric,
- TestTopicId::Numeric,
+ TestStreamId::Named,
+ TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestTopicUpdateCmd::new(
- 2,
+ 1,
String::from("production"),
- 3,
+ 0,
String::from("topic"),
Default::default(),
None,
@@ -274,14 +273,14 @@ pub async fn should_be_successful() {
MaxTopicSize::Unlimited,
1,
TestStreamId::Named,
- TestTopicId::Numeric,
+ TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestTopicUpdateCmd::new(
- 3,
+ 2,
String::from("testing"),
- 5,
+ 0,
String::from("development"),
Default::default(),
None,
@@ -292,15 +291,15 @@ pub async fn should_be_successful() {
None,
MaxTopicSize::Unlimited,
1,
- TestStreamId::Numeric,
+ TestStreamId::Named,
TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestTopicUpdateCmd::new(
- 2,
+ 3,
String::from("other"),
- 2,
+ 0,
String::from("probe"),
Default::default(),
None,
@@ -316,15 +315,15 @@ pub async fn should_be_successful() {
]),
MaxTopicSize::Unlimited,
1,
- TestStreamId::Numeric,
- TestTopicId::Numeric,
+ TestStreamId::Named,
+ TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestTopicUpdateCmd::new(
- 3,
+ 4,
String::from("stream"),
- 1,
+ 0,
String::from("testing"),
Default::default(),
Some(vec![String::from("1s")]),
@@ -335,15 +334,15 @@ pub async fn should_be_successful() {
Some(vec![String::from("1m 6s")]),
MaxTopicSize::ServerDefault,
1,
- TestStreamId::Numeric,
- TestTopicId::Numeric,
+ TestStreamId::Named,
+ TestTopicId::Named,
))
.await;
iggy_cmd_test
.execute_test(TestTopicUpdateCmd::new(
- 4,
+ 5,
String::from("testing"),
- 2,
+ 0,
String::from("testing"),
Default::default(),
Some(vec![
@@ -358,7 +357,7 @@ pub async fn should_be_successful() {
None,
MaxTopicSize::Unlimited,
1,
- TestStreamId::Numeric,
+ TestStreamId::Named,
TestTopicId::Named,
))
.await;
diff --git a/core/server/src/binary/handlers/topics/create_topic_handler.rs
b/core/server/src/binary/handlers/topics/create_topic_handler.rs
index 5e456e6a..2f1ee6cd 100644
--- a/core/server/src/binary/handlers/topics/create_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/create_topic_handler.rs
@@ -71,8 +71,7 @@ impl ServerCommandHandler for CreateTopic {
stream_id: self.stream_id.clone(),
topic,
};
- let _responses = shard.broadcast_event_to_all_shards(event).await;
-
+ let responses = shard.broadcast_event_to_all_shards(event).await;
let partitions = shard
.create_partitions2(
session,
@@ -86,7 +85,7 @@ impl ServerCommandHandler for CreateTopic {
topic_id: Identifier::numeric(topic_id as u32).unwrap(),
partitions,
};
- let _responses = shard.broadcast_event_to_all_shards(event).await;
+ let responses = shard.broadcast_event_to_all_shards(event).await;
let response = shard.streams2.with_topic_by_id(
&self.stream_id,
&Identifier::numeric(topic_id as u32).unwrap(),
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 8e6c06b3..250a52a5 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -235,12 +235,10 @@ pub fn create_shard_connections(
shards_set: &HashSet<usize>,
) -> (Vec<ShardConnector<ShardFrame>>, Vec<(u16, StopSender)>) {
let shards_count = shards_set.len();
- let mut shards_vec: Vec<usize> = shards_set.iter().cloned().collect();
- shards_vec.sort();
- let connectors: Vec<ShardConnector<ShardFrame>> = shards_vec
- .into_iter()
- .map(|id| ShardConnector::new(id as u16, shards_count))
+ // Create connectors with sequential IDs (0, 1, 2, ...) regardless of CPU
core numbers
+ let connectors: Vec<ShardConnector<ShardFrame>> = (0..shards_count)
+ .map(|idx| ShardConnector::new(idx as u16, shards_count))
.collect();
let shutdown_handles = connectors
diff --git a/core/server/src/http/http_server.rs
b/core/server/src/http/http_server.rs
index 61ebe091..9c3dd114 100644
--- a/core/server/src/http/http_server.rs
+++ b/core/server/src/http/http_server.rs
@@ -124,6 +124,15 @@ pub async fn start_http_server(
.expect("Failed to get local address for HTTP server");
info!("Started {api_name} on: {address}");
+ // Notify shard about the bound address
+ use crate::shard::transmission::event::ShardEvent;
+ use iggy_common::TransportProtocol;
+ let event = ShardEvent::AddressBound {
+ protocol: TransportProtocol::Http,
+ address,
+ };
+ shard.handle_event(event).await.ok();
+
let service =
app.into_make_service_with_connect_info::<CompioSocketAddr>();
// Spawn the server in a task so we can handle shutdown
@@ -155,6 +164,15 @@ pub async fn start_http_server(
info!("Started {api_name} on: {address}");
+ // Notify shard about the bound address
+ use crate::shard::transmission::event::ShardEvent;
+ use iggy_common::TransportProtocol;
+ let event = ShardEvent::AddressBound {
+ protocol: TransportProtocol::Http,
+ address,
+ };
+ shard.handle_event(event).await.ok();
+
let service = app.into_make_service_with_connect_info::<SocketAddr>();
// Create a handle for graceful shutdown
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index 37f225e6..36091a5a 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -276,7 +276,7 @@ async fn main() -> Result<(), ServerError> {
for (id, cpu_id) in shards_set
.into_iter()
.enumerate()
- .map(|(id, shard_id)| (id as u16, shard_id))
+ .map(|(idx, cpu)| (idx as u16, cpu))
{
let streams = streams.clone();
let shards_table = shards_table.clone();
diff --git a/core/server/src/quic/quic_server.rs
b/core/server/src/quic/quic_server.rs
index 6f29965c..d484a045 100644
--- a/core/server/src/quic/quic_server.rs
+++ b/core/server/src/quic/quic_server.rs
@@ -16,12 +16,13 @@
* under the License.
*/
-use std::fs::File;
-use std::io::BufReader;
-use std::net::SocketAddr;
-use std::rc::Rc;
-use std::sync::Arc;
-
+use crate::configs::quic::QuicConfig;
+use crate::quic::{COMPONENT, listener, quic_socket};
+use crate::server_error::QuicError;
+use crate::shard::IggyShard;
+use crate::shard::task_registry::ShutdownToken;
+use crate::shard::transmission::event::ShardEvent;
+use crate::shard_info;
use anyhow::Result;
use compio_quic::{
Endpoint, EndpointConfig, IdleTimeout, ServerBuilder, ServerConfig,
TransportConfig, VarInt,
@@ -29,13 +30,12 @@ use compio_quic::{
use error_set::ErrContext;
use rustls::crypto::ring::default_provider;
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
-use tracing::{error, info, trace, warn};
-
-use crate::configs::quic::QuicConfig;
-use crate::quic::{COMPONENT, listener, quic_socket};
-use crate::server_error::QuicError;
-use crate::shard::IggyShard;
-use crate::shard::task_registry::ShutdownToken;
+use std::fs::File;
+use std::io::BufReader;
+use std::net::SocketAddr;
+use std::rc::Rc;
+use std::sync::Arc;
+use tracing::{error, trace, warn};
/// Starts the QUIC server.
/// Returns the address the server is listening on.
@@ -58,13 +58,28 @@ pub async fn spawn_quic_server(
}
let config = shard.config.quic.clone();
- let addr: SocketAddr = config.address.parse().map_err(|e| {
+ let mut addr: SocketAddr = config.address.parse().map_err(|e| {
error!("Failed to parse QUIC address '{}': {}", config.address, e);
iggy_common::IggyError::QuicError
})?;
- info!(
+
+ if shard.id != 0 && addr.port() == 0 {
+ shard_info!(shard.id, "Waiting for QUIC address from shard 0...");
+ loop {
+ if let Some(bound_addr) = shard.quic_bound_address.get() {
+ addr = bound_addr;
+ shard_info!(shard.id, "Received QUIC address: {}", addr);
+ break;
+ }
+ compio::time::sleep(std::time::Duration::from_millis(10)).await;
+ }
+ }
+
+ shard_info!(
+ shard.id,
"Initializing Iggy QUIC server on shard {} for address {}",
- shard.id, addr
+ shard.id,
+ addr
);
let server_config = configure_quic(&config).map_err(|e| {
@@ -101,11 +116,28 @@ pub async fn spawn_quic_server(
iggy_common::IggyError::CannotBindToSocket(addr.to_string())
})?;
- info!(
- "Iggy QUIC server has started for shard {} on {}",
- shard.id, actual_addr
+ shard_info!(
+ shard.id,
+ "Iggy QUIC server has started on: {:?}",
+ actual_addr
);
- shard.quic_bound_address.set(Some(actual_addr));
+
+ if shard.id == 0 {
+ // Store bound address locally
+ shard.quic_bound_address.set(Some(actual_addr));
+
+ if addr.port() == 0 {
+ // Broadcast to other shards for SO_REUSEPORT binding
+ let event = ShardEvent::AddressBound {
+ protocol: iggy_common::TransportProtocol::Quic,
+ address: actual_addr,
+ };
+ shard.broadcast_event_to_all_shards(event).await;
+ }
+ } else {
+ shard.quic_bound_address.set(Some(actual_addr));
+ }
+
listener::start(endpoint, shard, shutdown).await
}
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index ce9cf7de..a439f7e3 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -16,6 +16,10 @@
* under the License.
*/
+use super::{
+ IggyShard, TaskRegistry, transmission::connector::ShardConnector,
+ transmission::frame::ShardFrame,
+};
use crate::{
configs::server::ServerConfig,
shard::{Shard, ShardInfo, namespace::IggyNamespace},
@@ -33,11 +37,6 @@ use std::{
sync::atomic::AtomicBool,
};
-use super::{
- IggyShard, TaskRegistry, transmission::connector::ShardConnector,
- transmission::frame::ShardFrame,
-};
-
#[derive(Default)]
pub struct IggyShardBuilder {
id: Option<u16>,
@@ -132,28 +131,37 @@ impl IggyShardBuilder {
let shards = connections.into_iter().map(Shard::new).collect();
// Initialize metrics
- let metrics = self.metrics.unwrap_or_else(|| Metrics::init());
+ let metrics = self.metrics.unwrap_or_else(Metrics::init);
// Create TaskRegistry for this shard
let task_registry = Rc::new(TaskRegistry::new(id));
+ // Create notification channel for config writer
+ let (config_writer_notify, config_writer_receiver) =
async_channel::bounded(1);
+
+ // Trigger initial check in case servers bind before task starts
+ let _ = config_writer_notify.try_send(());
+
IggyShard {
- id: id,
- shards: shards,
+ id,
+ shards,
shards_table,
streams2: streams, // TODO: Fixme
users: RefCell::new(users),
- encryptor: encryptor,
- config: config,
- version: version,
- state: state,
- stop_receiver: stop_receiver,
- stop_sender: stop_sender,
+ encryptor,
+ config,
+ version,
+ state,
+ stop_receiver,
+ stop_sender,
messages_receiver: Cell::new(Some(frame_receiver)),
- metrics: metrics,
+ metrics,
is_shutting_down: AtomicBool::new(false),
tcp_bound_address: Cell::new(None),
quic_bound_address: Cell::new(None),
+ http_bound_address: Cell::new(None),
+ config_writer_notify,
+ config_writer_receiver,
task_registry,
permissioner: Default::default(),
client_manager: Default::default(),
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 24c80a2b..f95a63d0 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -63,13 +63,14 @@ use crate::{
};
use ahash::{AHashMap, AHashSet, HashMap};
use builder::IggyShardBuilder;
+use compio::io::AsyncWriteAtExt;
use dashmap::DashMap;
use error_set::ErrContext;
use futures::future::try_join_all;
use hash32::{Hasher, Murmur3Hasher};
use iggy_common::{
- EncryptorKind, IdKind, Identifier, IggyError, IggyTimestamp, Permissions,
PollingKind, UserId,
- UserStatus,
+ EncryptorKind, IdKind, Identifier, IggyError, IggyTimestamp, Permissions,
PollingKind,
+ TransportProtocol, UserId, UserStatus,
defaults::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME},
locking::IggyRwLockFn,
};
@@ -164,6 +165,9 @@ pub struct IggyShard {
pub(crate) is_shutting_down: AtomicBool,
pub(crate) tcp_bound_address: Cell<Option<SocketAddr>>,
pub(crate) quic_bound_address: Cell<Option<SocketAddr>>,
+ pub(crate) http_bound_address: Cell<Option<SocketAddr>>,
+ config_writer_notify: async_channel::Sender<()>,
+ config_writer_receiver: async_channel::Receiver<()>,
pub(crate) task_registry: Rc<TaskRegistry>,
}
@@ -181,6 +185,13 @@ impl IggyShard {
fn init_tasks(self: &Rc<Self>) {
continuous::spawn_message_pump(self.clone());
+ // Spawn config writer task on shard 0 if we need to wait for bound
addresses
+ if self.id == 0
+ && (self.config.tcp.enabled || self.config.quic.enabled ||
self.config.http.enabled)
+ {
+ self.spawn_config_writer_task();
+ }
+
if self.config.tcp.enabled {
continuous::spawn_tcp_server(self.clone());
}
@@ -220,6 +231,106 @@ impl IggyShard {
}
}
+ fn spawn_config_writer_task(self: &Rc<Self>) {
+ let shard = self.clone();
+ let tcp_enabled = self.config.tcp.enabled;
+ let quic_enabled = self.config.quic.enabled;
+ let http_enabled = self.config.http.enabled;
+
+ let notify_receiver = shard.config_writer_receiver.clone();
+
+ self.task_registry
+ .oneshot("config_writer")
+ .critical(false)
+ .run(move |_shutdown| async move {
+ // Wait for notifications until all servers have bound
+ loop {
+ notify_receiver
+ .recv()
+ .await
+ .map_err(|_| IggyError::CannotWriteToFile)
+ .with_error_context(|_| {
+ "config_writer: notification channel closed before
all servers bound"
+ })?;
+
+ let tcp_ready = !tcp_enabled ||
shard.tcp_bound_address.get().is_some();
+ let quic_ready = !quic_enabled ||
shard.quic_bound_address.get().is_some();
+ let http_ready = !http_enabled ||
shard.http_bound_address.get().is_some();
+
+ if tcp_ready && quic_ready && http_ready {
+ break;
+ }
+ }
+
+ let mut current_config = shard.config.clone();
+
+ let tcp_addr = shard.tcp_bound_address.get();
+ let quic_addr = shard.quic_bound_address.get();
+ let http_addr = shard.http_bound_address.get();
+
+ shard_info!(
+ shard.id,
+ "Config writer: TCP addr = {:?}, QUIC addr = {:?}, HTTP
addr = {:?}",
+ tcp_addr,
+ quic_addr,
+ http_addr
+ );
+
+ if let Some(tcp_addr) = tcp_addr {
+ current_config.tcp.address = tcp_addr.to_string();
+ }
+
+ if let Some(quic_addr) = quic_addr {
+ current_config.quic.address = quic_addr.to_string();
+ }
+
+ if let Some(http_addr) = http_addr {
+ current_config.http.address = http_addr.to_string();
+ }
+
+ let runtime_path = current_config.system.get_runtime_path();
+ let config_path =
format!("{runtime_path}/current_config.toml");
+ let content = toml::to_string(¤t_config)
+ .map_err(|_| IggyError::CannotWriteToFile)
+ .with_error_context(|_| "config_writer: cannot serialize
current_config")?;
+
+ let mut file = compio::fs::OpenOptions::new()
+ .write(true)
+ .create(true)
+ .truncate(true)
+ .open(&config_path)
+ .await
+ .map_err(|_| IggyError::CannotWriteToFile)
+ .with_error_context(|_| {
+ format!("config_writer: failed to open current config
at {config_path}")
+ })?;
+
+ file.write_all_at(content.into_bytes(), 0)
+ .await
+ .0
+ .map_err(|_| IggyError::CannotWriteToFile)
+ .with_error_context(|_| {
+ format!("config_writer: failed to write current config
to {config_path}")
+ })?;
+
+ file.sync_all()
+ .await
+ .map_err(|_| IggyError::CannotWriteToFile)
+ .with_error_context(|_| {
+ format!("config_writer: failed to fsync current config
to {config_path}")
+ })?;
+
+ shard_info!(
+ shard.id,
+ "Current config written and synced to: {} with all bound
addresses",
+ config_path
+ );
+
+ Ok(())
+ })
+ .spawn();
+ }
+
pub async fn run(self: &Rc<Self>) -> Result<(), IggyError> {
let now = Instant::now();
@@ -481,7 +592,7 @@ impl IggyShard {
partition_id,
|(_, _, _, offset, _, _, _)| {
let current_offset =
offset.load(Ordering::Relaxed);
- let mut requested_count = 0;
+ let mut requested_count = count as u64;
if requested_count > current_offset + 1 {
requested_count = current_offset + 1
}
@@ -530,25 +641,38 @@ impl IggyShard {
),
};
- let Some(consumer_offset) = consumer_offset else {
- return
Err(IggyError::ConsumerOffsetNotFound(consumer_id));
+ let batches = if consumer_offset.is_none() {
+ let batches = self
+ .streams2
+ .get_messages_by_offset(
+ &stream_id,
+ &topic_id,
+ partition_id,
+ 0,
+ count,
+ )
+ .await?;
+ Ok(batches)
+ } else {
+ let consumer_offset = consumer_offset.unwrap();
+ let offset = consumer_offset + 1;
+ trace!(
+ "Getting next messages for consumer id: {} for
partition: {} from offset: {}...",
+ consumer_id, partition_id, offset
+ );
+ let batches = self
+ .streams2
+ .get_messages_by_offset(
+ &stream_id,
+ &topic_id,
+ partition_id,
+ offset,
+ count,
+ )
+ .await?;
+ Ok(batches)
};
- let offset = consumer_offset + 1;
- trace!(
- "Getting next messages for consumer id: {} for
partition: {} from offset: {}...",
- consumer_id, partition_id, offset
- );
- let batches = self
- .streams2
- .get_messages_by_offset(
- &stream_id,
- &topic_id,
- partition_id,
- offset,
- count,
- )
- .await?;
- Ok(batches)
+ batches
}
}?;
@@ -632,7 +756,7 @@ impl IggyShard {
}
}
- async fn handle_event(&self, event: ShardEvent) -> Result<(), IggyError> {
+ pub(crate) async fn handle_event(&self, event: ShardEvent) -> Result<(),
IggyError> {
match event {
ShardEvent::LoginUser {
client_id,
@@ -736,9 +860,37 @@ impl IggyShard {
self.update_permissions_bypass_auth(&user_id,
permissions.to_owned())?;
Ok(())
}
- ShardEvent::TcpBound { address } => {
- info!("Received TcpBound event with address: {}", address);
- self.tcp_bound_address.set(Some(address));
+ ShardEvent::AddressBound { protocol, address } => {
+ shard_info!(
+ self.id,
+ "Received AddressBound event for {:?} with address: {}",
+ protocol,
+ address
+ );
+ match protocol {
+ TransportProtocol::Tcp => {
+ self.tcp_bound_address.set(Some(address));
+ // Notify config writer that a server has bound
+ let _ = self.config_writer_notify.try_send(());
+ }
+ TransportProtocol::Quic => {
+ self.quic_bound_address.set(Some(address));
+ // Notify config writer that a server has bound
+ let _ = self.config_writer_notify.try_send(());
+ }
+ TransportProtocol::Http => {
+ self.http_bound_address.set(Some(address));
+ // Notify config writer that a server has bound
+ let _ = self.config_writer_notify.try_send(());
+ }
+ _ => {
+ shard_warn!(
+ self.id,
+ "Received AddressBound event for unsupported
protocol: {:?}",
+ protocol
+ );
+ }
+ }
Ok(())
}
ShardEvent::CreatedStream2 { id, stream } => {
@@ -1017,6 +1169,7 @@ impl IggyShard {
// TODO: Fixme, maybe we should send response_sender
// and propagate errors back.
let event = event.clone();
+ /*
if matches!(
&event,
ShardEvent::CreatedStream2 { .. }
@@ -1030,13 +1183,16 @@ impl IggyShard {
| ShardEvent::CreatedPersonalAccessToken { .. }
| ShardEvent::DeletedConsumerGroup2 { .. }
) {
- let (sender, receiver) = async_channel::bounded(1);
- conn.send(ShardFrame::new(event.into(),
Some(sender.clone())));
- Some(receiver.clone())
+ */
+ let (sender, receiver) = async_channel::bounded(1);
+ conn.send(ShardFrame::new(event.into(), Some(sender.clone())));
+ Some(receiver.clone())
+ /*
} else {
conn.send(ShardFrame::new(event.into(), None));
None
}
+ */
})
{
match maybe_receiver {
diff --git a/core/server/src/shard/system/messages.rs
b/core/server/src/shard/system/messages.rs
index 3f8fd41d..7cb49a7e 100644
--- a/core/server/src/shard/system/messages.rs
+++ b/core/server/src/shard/system/messages.rs
@@ -56,10 +56,7 @@ impl IggyShard {
messages_key: &[u8],
) -> usize {
let messages_key_hash = hash::calculate_32(messages_key) as usize;
- let mut partition_id = messages_key_hash % upperbound;
- if partition_id == 0 {
- partition_id = upperbound;
- }
+ let partition_id = messages_key_hash % upperbound;
shard_trace!(
shard_id,
"Calculated partition ID: {} for messages key: {:?}, hash: {}",
diff --git a/core/server/src/shard/system/partitions.rs
b/core/server/src/shard/system/partitions.rs
index d78192f0..3e086c5a 100644
--- a/core/server/src/shard/system/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -212,6 +212,7 @@ impl IggyShard {
let numeric_topic_id =
self.streams2
.with_topic_by_id(stream_id, topic_id,
topics::helpers::get_topic_id());
+ let shards_count = self.get_available_shards_count();
for partition in partitions {
let actual_id = partition.id();
let id = self.streams2.with_partitions_mut(
@@ -224,9 +225,9 @@ impl IggyShard {
"create_partitions_bypass_auth: partition mismatch ID, wrong
creation order ?!"
);
let ns = IggyNamespace::new(numeric_stream_id, numeric_topic_id,
id);
- let shard_info = self
- .find_shard_table_record(&ns)
- .expect("create_partitions_bypass_auth: missing shard table
record");
+ let shard_id = crate::shard::calculate_shard_assignment(&ns,
shards_count);
+ let shard_info = ShardInfo::new(shard_id);
+ self.insert_shard_table_record(ns, shard_info);
if self.id == shard_info.id {
self.init_log(stream_id, topic_id, id).await?;
}
@@ -243,13 +244,15 @@ impl IggyShard {
partitions_count: u32,
) -> Result<Vec<usize>, IggyError> {
self.ensure_authenticated(session)?;
+ self.ensure_partitions_exist(stream_id, topic_id, partitions_count)?;
+
let numeric_stream_id = self
.streams2
.with_stream_by_id(stream_id, streams::helpers::get_stream_id());
let numeric_topic_id =
self.streams2
.with_topic_by_id(stream_id, topic_id,
topics::helpers::get_topic_id());
- // Claude garbage, rework this.
+
self.validate_partition_permissions(
session,
numeric_stream_id as u32,
@@ -263,7 +266,6 @@ impl IggyShard {
.map(|p| p.stats().parent().clone())
.expect("delete_partitions: no partitions to deletion");
// Reassign the partitions count as it could get clamped by the
`delete_partitions_base2` method.
- let partitions_count = partitions.len() as u32;
let mut deleted_ids = Vec::with_capacity(partitions.len());
let mut total_messages_count = 0;
@@ -335,7 +337,11 @@ impl IggyShard {
partitions_count: u32,
partition_ids: Vec<usize>,
) -> Result<(), IggyError> {
- assert_eq!(partitions_count as usize, partition_ids.len());
+ self.ensure_partitions_exist(stream_id, topic_id, partitions_count)?;
+
+ if partitions_count as usize != partition_ids.len() {
+ return Err(IggyError::InvalidPartitionsCount);
+ }
let partitions = self.delete_partitions_base2(stream_id, topic_id,
partitions_count);
for (deleted_partition_id, actual_deleted_partition_id) in partitions
diff --git a/core/server/src/shard/system/snapshot/mod.rs
b/core/server/src/shard/system/snapshot/mod.rs
index 630dd2be..0c95de91 100644
--- a/core/server/src/shard/system/snapshot/mod.rs
+++ b/core/server/src/shard/system/snapshot/mod.rs
@@ -67,26 +67,26 @@ impl IggyShard {
let now = Instant::now();
for snapshot_type in snapshot_types {
+ info!("Processing snapshot type: {:?}", snapshot_type);
match get_command_result(snapshot_type,
self.config.system.clone()).await {
Ok(temp_file) => {
+ info!(
+ "Got temp file for {:?}: {}",
+ snapshot_type,
+ temp_file.path().display()
+ );
let filename = format!("{snapshot_type}.txt");
let entry = ZipEntryBuilder::new(filename.clone().into(),
compression);
- let file = OpenOptions::new()
- .read(true)
- .open(temp_file.path())
- .await
- .map_err(|e| {
- error!("Failed to open temporary file: {}", e);
- IggyError::SnapshotFileCompletionFailed
- })?;
-
- let content = Vec::new();
- let (result, content) = file.read_exact_at(content,
0).await.into();
- if let Err(e) = result {
- error!("Failed to read temporary file: {}", e);
- continue;
- }
+ // Read file using compio fs
+ let content = match
compio::fs::read(temp_file.path()).await {
+ Ok(data) => data,
+ Err(e) => {
+ error!("Failed to read temporary file: {}", e);
+ continue;
+ }
+ };
+
info!(
"Read {} bytes from temp file for {}",
content.len(),
@@ -129,19 +129,41 @@ async fn write_command_output_to_temp_file(
command: &mut Command,
) -> Result<NamedTempFile, std::io::Error> {
let output = command.output()?;
+
+ info!(
+ "Command output: {} bytes, stderr: {}",
+ output.stdout.len(),
+ String::from_utf8_lossy(&output.stderr)
+ );
+
let temp_file = NamedTempFile::new()?;
+
+ // Use compio to write the file - create/truncate to ensure clean write
let mut file = OpenOptions::new()
+ .create(true)
.write(true)
+ .truncate(true)
.open(temp_file.path())
.await?;
- let (result, _) = file.write_all_at(output.stdout, 0).await.into();
+
+ // Write the command output - compio takes ownership of the buffer
+ let stdout = output.stdout;
+ let (result, _buf) = file.write_all_at(stdout, 0).await.into();
result?;
+
file.sync_all().await?;
+
+ info!(
+ "Wrote {} bytes to temp file: {}",
+ _buf.len(),
+ temp_file.path().display()
+ );
+
Ok(temp_file)
}
async fn get_filesystem_overview() -> Result<NamedTempFile, std::io::Error> {
- write_command_output_to_temp_file(Command::new("ls").args(["-la", "/tmp",
"/proc"])).await
+ write_command_output_to_temp_file(&mut Command::new("ls").args(["-la",
"/tmp", "/proc"])).await
}
async fn get_process_info() -> Result<NamedTempFile, std::io::Error> {
@@ -190,7 +212,7 @@ async fn get_resource_usage() -> Result<NamedTempFile,
std::io::Error> {
}
async fn get_test_snapshot() -> Result<NamedTempFile, std::io::Error> {
- write_command_output_to_temp_file(Command::new("echo").arg("test")).await
+ write_command_output_to_temp_file(&mut
Command::new("echo").arg("test")).await
}
async fn get_server_logs(config: Arc<SystemConfig>) -> Result<NamedTempFile,
std::io::Error> {
diff --git a/core/server/src/shard/system/utils.rs
b/core/server/src/shard/system/utils.rs
index 0c9b637b..d2518da7 100644
--- a/core/server/src/shard/system/utils.rs
+++ b/core/server/src/shard/system/utils.rs
@@ -50,6 +50,24 @@ impl IggyShard {
Ok(())
}
+ pub fn ensure_partitions_exist(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partitions_count: u32,
+ ) -> Result<(), IggyError> {
+ self.ensure_topic_exists(stream_id, topic_id)?;
+ let actual_partitions_count =
+ self.streams2
+ .with_partitions(stream_id, topic_id, |partitions|
partitions.len());
+
+ if partitions_count > actual_partitions_count as u32 {
+ return Err(IggyError::InvalidPartitionsCount);
+ }
+
+ Ok(())
+ }
+
pub fn resolve_consumer_with_partition_id(
&self,
stream_id: &Identifier,
diff --git a/core/server/src/shard/transmission/event.rs
b/core/server/src/shard/transmission/event.rs
index 544c6871..61bd5647 100644
--- a/core/server/src/shard/transmission/event.rs
+++ b/core/server/src/shard/transmission/event.rs
@@ -159,7 +159,8 @@ pub enum ShardEvent {
topic_id: Identifier,
group_id: Identifier,
},
- TcpBound {
+ AddressBound {
+ protocol: TransportProtocol,
address: SocketAddr,
},
}
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index 6167987c..5d3acc83 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -381,7 +381,10 @@ impl MainOps for Streams {
};
let Some(consumer_offset) = consumer_offset else {
- return Err(IggyError::ConsumerOffsetNotFound(consumer_id));
+ let batches = self
+ .get_messages_by_offset(stream_id, topic_id,
partition_id, 0, count)
+ .await?;
+ return Ok((metadata, batches));
};
let offset = consumer_offset + 1;
trace!(
@@ -607,6 +610,10 @@ impl Streams {
let mut current_offset = offset;
for idx in range {
+ if remaining_count == 0 {
+ break;
+ }
+
let (segment_start_offset, segment_end_offset) =
self.with_partition_by_id(
stream_id,
topic_id,
@@ -658,10 +665,6 @@ impl Streams {
}
batches.add_batch_set(messages);
-
- if remaining_count == 0 {
- break;
- }
}
Ok(batches)
@@ -890,6 +893,10 @@ impl Streams {
let mut batches = IggyMessagesBatchSet::empty();
for idx in range {
+ if remaining_count == 0 {
+ break;
+ }
+
let segment_end_timestamp = self.with_partition_by_id(
stream_id,
topic_id,
@@ -922,10 +929,6 @@ impl Streams {
remaining_count = remaining_count.saturating_sub(messages_count);
batches.add_batch_set(messages);
-
- if remaining_count == 0 {
- break;
- }
}
Ok(batches)
diff --git a/core/server/src/streaming/topics/topic2.rs
b/core/server/src/streaming/topics/topic2.rs
index e46a5c3b..2142b19b 100644
--- a/core/server/src/streaming/topics/topic2.rs
+++ b/core/server/src/streaming/topics/topic2.rs
@@ -22,8 +22,8 @@ pub struct TopicAuxilary {
impl TopicAuxilary {
pub fn get_next_partition_id(&self, shard_id: u16, upperbound: usize) ->
usize {
let mut partition_id = self.current_partition_id.fetch_add(1,
Ordering::AcqRel);
- if partition_id > upperbound {
- partition_id = 1;
+ if partition_id >= upperbound {
+ partition_id = 0;
self.current_partition_id
.swap(partition_id + 1, Ordering::Release);
}
diff --git a/core/server/src/tcp/tcp_listener.rs
b/core/server/src/tcp/tcp_listener.rs
index d3aa84a6..0614cb01 100644
--- a/core/server/src/tcp/tcp_listener.rs
+++ b/core/server/src/tcp/tcp_listener.rs
@@ -102,35 +102,17 @@ pub async fn start(
);
if shard.id == 0 {
+ // Store bound address locally
+ shard.tcp_bound_address.set(Some(actual_addr));
+
if addr.port() == 0 {
- let event = ShardEvent::TcpBound {
+ // Broadcast to other shards for SO_REUSEPORT binding
+ let event = ShardEvent::AddressBound {
+ protocol: TransportProtocol::Tcp,
address: actual_addr,
};
shard.broadcast_event_to_all_shards(event).await;
}
-
- let mut current_config = shard.config.clone();
- current_config.tcp.address = actual_addr.to_string();
-
- let runtime_path = current_config.system.get_runtime_path();
- let current_config_path =
format!("{runtime_path}/current_config.toml");
- let current_config_content =
- toml::to_string(¤t_config).expect("Cannot serialize
current_config");
-
- let buf_result = compio::fs::write(¤t_config_path,
current_config_content).await;
- match buf_result.0 {
- Ok(_) => shard_info!(
- shard.id,
- "Current config written to: {}",
- current_config_path
- ),
- Err(e) => shard_error!(
- shard.id,
- "Failed to write current config to {}: {}",
- current_config_path,
- e
- ),
- }
}
accept_loop(server_name, listener, shard, shutdown).await
diff --git a/core/server/src/tcp/tcp_tls_listener.rs
b/core/server/src/tcp/tcp_tls_listener.rs
index dd028fb9..1bf22e03 100644
--- a/core/server/src/tcp/tcp_tls_listener.rs
+++ b/core/server/src/tcp/tcp_tls_listener.rs
@@ -72,35 +72,17 @@ pub(crate) async fn start(
//TODO: Fix me, this needs to take into account that first shard id
potentially can be greater than 0.
if shard.id == 0 {
+ // Store bound address locally
+ shard.tcp_bound_address.set(Some(actual_addr));
+
if addr.port() == 0 {
- let event = ShardEvent::TcpBound {
+ // Broadcast to other shards for SO_REUSEPORT binding
+ let event = ShardEvent::AddressBound {
+ protocol: TransportProtocol::Tcp,
address: actual_addr,
};
shard.broadcast_event_to_all_shards(event).await;
}
-
- let mut current_config = shard.config.clone();
- current_config.tcp.address = actual_addr.to_string();
-
- let runtime_path = current_config.system.get_runtime_path();
- let current_config_path =
format!("{runtime_path}/current_config.toml");
- let current_config_content =
- toml::to_string(¤t_config).expect("Cannot serialize
current_config");
-
- let buf_result = compio::fs::write(¤t_config_path,
current_config_content).await;
- match buf_result.0 {
- Ok(_) => shard_info!(
- shard.id,
- "Current config written to: {}",
- current_config_path
- ),
- Err(e) => shard_error!(
- shard.id,
- "Failed to write current config to {}: {}",
- current_config_path,
- e
- ),
- }
}
// Ensure rustls crypto provider is installed
