You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/08/06 09:25:45 UTC
(iggy) branch config_path_improvements created (now 516d6179)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch config_path_improvements
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 516d6179 feat(io_uring): create topic directories on topic creation and improve config path resolver
This branch includes the following new commits:
new 516d6179 feat(io_uring): create topic directories on topic creation and improve config path resolver
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: feat(io_uring): create topic directories on topic creation and improve config path resolver
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch config_path_improvements
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 516d6179b1b2559ee3594e9291d048a9040a9122
Author: numminex <g....@gmail.com>
AuthorDate: Wed Aug 6 11:25:34 2025 +0200
feat(io_uring): create topic directories on topic creation and improve config path resolver
---
.../tests/streaming/common/test_setup.rs | 20 ++++----
core/integration/tests/streaming/segment.rs | 14 +++---
core/integration/tests/streaming/topic.rs | 8 ++--
core/server/src/configs/system.rs | 40 ++++++++++------
core/server/src/shard/system/streams.rs | 2 +-
core/server/src/shard/system/topics.rs | 1 +
core/server/src/slab/mod.rs | 2 +-
core/server/src/streaming/partitions/partition.rs | 28 +++++------
core/server/src/streaming/segments/segment.rs | 20 ++++----
core/server/src/streaming/streams/storage2.rs | 18 +++-----
core/server/src/streaming/streams/stream.rs | 10 ++--
core/server/src/streaming/topics/mod.rs | 1 +
core/server/src/streaming/topics/storage2.rs | 54 ++++++++++++++++++++++
core/server/src/streaming/topics/topic.rs | 14 +++---
core/server/src/streaming/topics/topic2.rs | 8 ++--
15 files changed, 151 insertions(+), 89 deletions(-)
diff --git a/core/integration/tests/streaming/common/test_setup.rs b/core/integration/tests/streaming/common/test_setup.rs
index 88c32194..bfa2a128 100644
--- a/core/integration/tests/streaming/common/test_setup.rs
+++ b/core/integration/tests/streaming/common/test_setup.rs
@@ -61,11 +61,11 @@ impl TestSetup {
pub async fn create_stream_directory(&self, stream_id: u32) {
self.create_streams_directory().await;
- if fs::metadata(&self.config.get_stream_path(stream_id))
+ if fs::metadata(&self.config.get_stream_path(stream_id as usize))
.await
.is_err()
{
- fs::create_dir(&self.config.get_stream_path(stream_id))
+ fs::create_dir(&self.config.get_stream_path(stream_id as usize))
.await
.unwrap();
}
@@ -73,11 +73,11 @@ impl TestSetup {
pub async fn create_topics_directory(&self, stream_id: u32) {
self.create_stream_directory(stream_id).await;
- if fs::metadata(&self.config.get_topics_path(stream_id))
+ if fs::metadata(&self.config.get_topics_path(stream_id as usize))
.await
.is_err()
{
- fs::create_dir(&self.config.get_topics_path(stream_id))
+ fs::create_dir(&self.config.get_topics_path(stream_id as usize))
.await
.unwrap();
}
@@ -85,11 +85,11 @@ impl TestSetup {
pub async fn create_topic_directory(&self, stream_id: u32, topic_id: u32) {
self.create_topics_directory(stream_id).await;
- if fs::metadata(&self.config.get_topic_path(stream_id, topic_id))
+ if fs::metadata(&self.config.get_topic_path(stream_id as usize, topic_id as usize))
.await
.is_err()
{
- fs::create_dir(&self.config.get_topic_path(stream_id, topic_id))
+ fs::create_dir(&self.config.get_topic_path(stream_id as usize, topic_id as usize))
.await
.unwrap();
}
@@ -97,11 +97,11 @@ impl TestSetup {
pub async fn create_partitions_directory(&self, stream_id: u32, topic_id: u32) {
self.create_topic_directory(stream_id, topic_id).await;
- if fs::metadata(&self.config.get_partitions_path(stream_id, topic_id))
+ if fs::metadata(&self.config.get_partitions_path(stream_id as usize, topic_id as usize))
.await
.is_err()
{
- fs::create_dir(&self.config.get_partitions_path(stream_id, topic_id))
+ fs::create_dir(&self.config.get_partitions_path(stream_id as usize, topic_id as usize))
.await
.unwrap();
}
@@ -117,7 +117,7 @@ impl TestSetup {
if fs::metadata(
&self
.config
- .get_partition_path(stream_id, topic_id, partition_id),
+ .get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize),
)
.await
.is_err()
@@ -125,7 +125,7 @@ impl TestSetup {
fs::create_dir(
&self
.config
- .get_partition_path(stream_id, topic_id, partition_id),
+ .get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize),
)
.await
.unwrap();
diff --git a/core/integration/tests/streaming/segment.rs b/core/integration/tests/streaming/segment.rs
index 4e097b8f..3e3bc7cf 100644
--- a/core/integration/tests/streaming/segment.rs
+++ b/core/integration/tests/streaming/segment.rs
@@ -68,7 +68,7 @@ async fn should_persist_segment() {
assert_persisted_segment(
&setup
.config
- .get_partition_path(stream_id, topic_id, partition_id),
+ .get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize),
start_offset,
)
.await;
@@ -105,7 +105,7 @@ async fn should_load_existing_segment_from_disk() {
assert_persisted_segment(
&setup
.config
- .get_partition_path(stream_id, topic_id, partition_id),
+ .get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize),
start_offset,
)
.await;
@@ -175,7 +175,7 @@ async fn should_persist_and_load_segment_with_messages() {
assert_persisted_segment(
&setup
.config
- .get_partition_path(stream_id, topic_id, partition_id),
+ .get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize),
start_offset,
)
.await;
@@ -261,7 +261,7 @@ async fn given_at_least_one_not_expired_message_segment_should_not_be_expired()
assert_persisted_segment(
&setup
.config
- .get_partition_path(stream_id, topic_id, partition_id),
+ .get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize),
start_offset,
)
.await;
@@ -408,9 +408,9 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
drop(stream);
let partition_path = setup.config.get_partition_path(
- stream_id.get_u32_value()?,
- topic_id.get_u32_value()?,
- partition_id,
+ stream_id.get_u32_value()? as usize,
+ topic_id.get_u32_value()? as usize,
+ partition_id as usize,
);
let initial_segments = get_segment_paths_for_partition(&partition_path);
diff --git a/core/integration/tests/streaming/topic.rs b/core/integration/tests/streaming/topic.rs
index 37c4560b..cf5ab718 100644
--- a/core/integration/tests/streaming/topic.rs
+++ b/core/integration/tests/streaming/topic.rs
@@ -60,7 +60,7 @@ async fn should_persist_topics_with_partitions_directories_and_info_file() {
assert_persisted_topic(
&topic.path,
- &setup.config.get_partitions_path(stream_id, topic_id),
+ &setup.config.get_partitions_path(stream_id as usize, topic_id as usize),
3,
)
.await;
@@ -97,7 +97,7 @@ async fn should_load_existing_topic_from_disk() {
topic.persist().await.unwrap();
assert_persisted_topic(
&topic.path,
- &setup.config.get_partitions_path(stream_id, topic_id),
+ &setup.config.get_partitions_path(stream_id as usize, topic_id as usize),
partitions_count,
)
.await;
@@ -174,7 +174,7 @@ async fn should_delete_existing_topic_from_disk() {
topic.persist().await.unwrap();
assert_persisted_topic(
&topic.path,
- &setup.config.get_partitions_path(stream_id, topic_id),
+ &setup.config.get_partitions_path(stream_id as usize, topic_id as usize),
partitions_count,
)
.await;
@@ -214,7 +214,7 @@ async fn should_purge_existing_topic_on_disk() {
topic.persist().await.unwrap();
assert_persisted_topic(
&topic.path,
- &setup.config.get_partitions_path(stream_id, topic_id),
+ &setup.config.get_partitions_path(stream_id as usize, topic_id as usize),
partitions_count,
)
.await;
diff --git a/core/server/src/configs/system.rs b/core/server/src/configs/system.rs
index 54a277cb..c734f004 100644
--- a/core/server/src/configs/system.rs
+++ b/core/server/src/configs/system.rs
@@ -193,19 +193,19 @@ impl SystemConfig {
format!("{}/{}", self.get_system_path(), self.stream.path)
}
- pub fn get_stream_path(&self, stream_id: u32) -> String {
+ pub fn get_stream_path(&self, stream_id: usize) -> String {
format!("{}/{}", self.get_streams_path(), stream_id)
}
- pub fn get_topics_path(&self, stream_id: u32) -> String {
+ pub fn get_topics_path(&self, stream_id: usize) -> String {
format!("{}/{}", self.get_stream_path(stream_id), self.topic.path)
}
- pub fn get_topic_path(&self, stream_id: u32, topic_id: u32) -> String {
+ pub fn get_topic_path(&self, stream_id: usize, topic_id: usize) -> String {
format!("{}/{}", self.get_topics_path(stream_id), topic_id)
}
- pub fn get_partitions_path(&self, stream_id: u32, topic_id: u32) -> String {
+ pub fn get_partitions_path(&self, stream_id: usize, topic_id: usize) -> String {
format!(
"{}/{}",
self.get_topic_path(stream_id, topic_id),
@@ -213,7 +213,12 @@ impl SystemConfig {
)
}
- pub fn get_partition_path(&self, stream_id: u32, topic_id: u32, partition_id: u32) -> String {
+ pub fn get_partition_path(
+ &self,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ ) -> String {
format!(
"{}/{}",
self.get_partitions_path(stream_id, topic_id),
@@ -221,7 +226,12 @@ impl SystemConfig {
)
}
- pub fn get_offsets_path(&self, stream_id: u32, topic_id: u32, partition_id: u32) -> String {
+ pub fn get_offsets_path(
+ &self,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ ) -> String {
format!(
"{}/offsets",
self.get_partition_path(stream_id, topic_id, partition_id)
@@ -230,9 +240,9 @@ impl SystemConfig {
pub fn get_consumer_offsets_path(
&self,
- stream_id: u32,
- topic_id: u32,
- partition_id: u32,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
) -> String {
format!(
"{}/consumers",
@@ -242,9 +252,9 @@ impl SystemConfig {
pub fn get_consumer_group_offsets_path(
&self,
- stream_id: u32,
- topic_id: u32,
- partition_id: u32,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
) -> String {
format!(
"{}/groups",
@@ -254,9 +264,9 @@ impl SystemConfig {
pub fn get_segment_path(
&self,
- stream_id: u32,
- topic_id: u32,
- partition_id: u32,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
start_offset: u64,
) -> String {
format!(
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index dc7cc8b9..1c8c7df6 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -218,7 +218,7 @@ impl IggyShard {
.borrow()
.create_stream(session.get_user_id())?;
let stream_id = self.create_stream2_base(name, stats).await?;
- create_stream_file_hierarchy(stream_id, &self.config.system).await?;
+ create_stream_file_hierarchy(self.id, stream_id, &self.config.system).await?;
Ok(stream_id)
}
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index da411296..46c6b5d2 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -192,6 +192,7 @@ impl IggyShard {
);
// TODO: Create dir hierarchy for the topic.
+
let partition_ids = self.create_partitions2_base(
stream_id,
&Identifier::numeric(topic_id as u32).unwrap(),
diff --git a/core/server/src/slab/mod.rs b/core/server/src/slab/mod.rs
index 39d0c943..b03c71ad 100644
--- a/core/server/src/slab/mod.rs
+++ b/core/server/src/slab/mod.rs
@@ -21,7 +21,7 @@ pub struct IndexedSlab<T: Keyed> {
index: AHashMap<T::Key, usize>,
}
-impl<T: Keyed + Default + Debug> IndexedSlab<T> {
+impl<T: Keyed + Debug> IndexedSlab<T> {
pub fn with_capacity(capacity: usize) -> Self {
Self {
slab: Slab::with_capacity(capacity),
diff --git a/core/server/src/streaming/partitions/partition.rs b/core/server/src/streaming/partitions/partition.rs
index fc346693..3103e2af 100644
--- a/core/server/src/streaming/partitions/partition.rs
+++ b/core/server/src/streaming/partitions/partition.rs
@@ -103,12 +103,12 @@ impl Partition {
segments_count_of_parent_stream: Arc<AtomicU32>,
created_at: IggyTimestamp,
) -> Partition {
- let partition_path = config.get_partition_path(stream_id, topic_id, partition_id);
- let offsets_path = config.get_offsets_path(stream_id, topic_id, partition_id);
+ let partition_path = config.get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize);
+ let offsets_path = config.get_offsets_path(stream_id as usize, topic_id as usize, partition_id as usize);
let consumer_offsets_path =
- config.get_consumer_offsets_path(stream_id, topic_id, partition_id);
+ config.get_consumer_offsets_path(stream_id as usize, topic_id as usize, partition_id as usize);
let consumer_group_offsets_path =
- config.get_consumer_group_offsets_path(stream_id, topic_id, partition_id);
+ config.get_consumer_group_offsets_path(stream_id as usize, topic_id as usize, partition_id as usize);
let message_deduplicator = match config.message_deduplication.enabled {
true => Some(MessageDeduplicator::new(
@@ -240,16 +240,16 @@ mod tests {
));
MemoryPool::init_pool(config.clone());
- let stream_id = 1;
- let topic_id = 2;
- let partition_id = 3;
+ let stream_id = 1usize;
+ let topic_id = 2usize;
+ let partition_id = 3usize;
let with_segment = true;
- let path = config.get_partition_path(stream_id, topic_id, partition_id);
+ let path = config.get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize);
let message_expiry = IggyExpiry::ExpireDuration(IggyDuration::from(10));
let partition = Partition::create(
- stream_id,
- topic_id,
- partition_id,
+ stream_id as u32,
+ topic_id as u32,
+ partition_id as u32,
with_segment,
config,
storage,
@@ -262,9 +262,9 @@ mod tests {
IggyTimestamp::now(),
);
- assert_eq!(partition.stream_id, stream_id);
- assert_eq!(partition.topic_id, topic_id);
- assert_eq!(partition.partition_id, partition_id);
+ assert_eq!(partition.stream_id, stream_id as u32);
+ assert_eq!(partition.topic_id, topic_id as u32);
+ assert_eq!(partition.partition_id, partition_id as u32);
assert_eq!(partition.partition_path, path);
assert_eq!(partition.current_offset, 0);
assert_eq!(partition.unsaved_messages_count, 0);
diff --git a/core/server/src/streaming/segments/segment.rs b/core/server/src/streaming/segments/segment.rs
index a3ac8588..b197cb41 100644
--- a/core/server/src/streaming/segments/segment.rs
+++ b/core/server/src/streaming/segments/segment.rs
@@ -85,7 +85,7 @@ impl Segment {
messages_count_of_parent_partition: Arc<AtomicU64>,
fresh: bool, // `fresh` means created and persisted in this runtime, in other words it's set to false when loading from disk
) -> Segment {
- let path = config.get_segment_path(stream_id, topic_id, partition_id, start_offset);
+ let path = config.get_segment_path(stream_id as usize, topic_id as usize, partition_id as usize, start_offset);
let messages_path = Self::get_messages_file_path(&path);
let index_path = Self::get_index_path(&path);
let message_expiry = match message_expiry {
@@ -460,9 +460,9 @@ mod tests {
#[tokio::test]
async fn should_be_created_given_valid_parameters() {
- let stream_id = 1;
- let topic_id = 2;
- let partition_id = 3;
+ let stream_id = 1usize;
+ let topic_id = 2usize;
+ let partition_id = 3usize;
let start_offset = 0;
let config = Arc::new(SystemConfig::default());
let path = config.get_segment_path(stream_id, topic_id, partition_id, start_offset);
@@ -478,9 +478,9 @@ mod tests {
MemoryPool::init_pool(config.clone());
let segment = Segment::create(
- stream_id,
- topic_id,
- partition_id,
+ stream_id as u32,
+ topic_id as u32,
+ partition_id as u32,
start_offset,
config,
message_expiry,
@@ -493,9 +493,9 @@ mod tests {
true,
);
- assert_eq!(segment.stream_id, stream_id);
- assert_eq!(segment.topic_id, topic_id);
- assert_eq!(segment.partition_id, partition_id);
+ assert_eq!(segment.stream_id, stream_id as u32);
+ assert_eq!(segment.topic_id, topic_id as u32);
+ assert_eq!(segment.partition_id, partition_id as u32);
assert_eq!(segment.start_offset(), start_offset);
assert_eq!(segment.end_offset(), 0);
assert_eq!(segment.get_messages_size(), 0);
diff --git a/core/server/src/streaming/streams/storage2.rs b/core/server/src/streaming/streams/storage2.rs
index bdbdacb4..658f2921 100644
--- a/core/server/src/streaming/streams/storage2.rs
+++ b/core/server/src/streaming/streams/storage2.rs
@@ -1,28 +1,22 @@
+use crate::{configs::system::SystemConfig, shard_info};
use compio::fs::create_dir_all;
use iggy_common::IggyError;
use std::path::Path;
-use crate::configs::system::SystemConfig;
-
pub async fn create_stream_file_hierarchy(
+ shard_id: u16,
id: usize,
config: &SystemConfig,
) -> Result<(), IggyError> {
- let id = id as u32;
let path = config.get_stream_path(id);
- let topics_path = config.get_topics_path(id);
if !Path::new(&path).exists() && create_dir_all(&path).await.is_err() {
- return Err(IggyError::CannotCreateStreamDirectory(id, path.clone()));
- }
-
- if !Path::new(&topics_path).exists() && create_dir_all(&topics_path).await.is_err() {
- return Err(IggyError::CannotCreateTopicsDirectory(
- id,
- topics_path.clone(),
+ return Err(IggyError::CannotCreateStreamDirectory(
+ id as u32,
+ path.clone(),
));
}
- tracing::info!("Saved stream with ID: {}.", id);
+ shard_info!(shard_id, "Saved stream with ID: {}.", id);
Ok(())
}
diff --git a/core/server/src/streaming/streams/stream.rs b/core/server/src/streaming/streams/stream.rs
index 8b59ad45..646fff6d 100644
--- a/core/server/src/streaming/streams/stream.rs
+++ b/core/server/src/streaming/streams/stream.rs
@@ -61,8 +61,8 @@ impl Stream {
config: Arc<SystemConfig>,
storage: Rc<SystemStorage>,
) -> Self {
- let path = config.get_stream_path(id);
- let topics_path = config.get_topics_path(id);
+ let path = config.get_stream_path(id as usize);
+ let topics_path = config.get_topics_path(id as usize);
Stream {
stream_id: id,
@@ -116,11 +116,11 @@ mod tests {
Arc::new(PersisterKind::FileWithSync(FileWithSyncPersister {})),
));
MemoryPool::init_pool(config.clone());
- let id = 1;
+ let id = 1u32;
let name = "test";
let config = Arc::new(SystemConfig::default());
- let path = config.get_stream_path(id);
- let topics_path = config.get_topics_path(id);
+ let path = config.get_stream_path(id as usize);
+ let topics_path = config.get_topics_path(id as usize);
let stream = Stream::create(id, name, config, storage);
diff --git a/core/server/src/streaming/topics/mod.rs b/core/server/src/streaming/topics/mod.rs
index 56e67d35..0e7bea60 100644
--- a/core/server/src/streaming/topics/mod.rs
+++ b/core/server/src/streaming/topics/mod.rs
@@ -25,6 +25,7 @@ pub mod partitions;
pub mod persistence;
pub mod segments;
pub mod storage;
+pub mod storage2;
pub mod topic;
pub mod topic2;
diff --git a/core/server/src/streaming/topics/storage2.rs b/core/server/src/streaming/topics/storage2.rs
new file mode 100644
index 00000000..ccc8f164
--- /dev/null
+++ b/core/server/src/streaming/topics/storage2.rs
@@ -0,0 +1,54 @@
+use super::COMPONENT;
+use compio::fs::create_dir_all;
+use iggy_common::IggyError;
+use std::path::Path;
+
+use crate::{configs::system::SystemConfig, shard_info, streaming::topics::topic2};
+
+async fn create_topic_file_hierarchy(
+ shard_id: u16,
+ stream_id: usize,
+ topic: &topic2::Topic,
+ config: &SystemConfig,
+) -> Result<(), IggyError> {
+ let topic_path = config.get_topic_path(stream_id, topic.id());
+ if !Path::new(&topic_path).exists() && create_dir_all(&topic_path).await.is_err() {
+ return Err(IggyError::CannotCreateTopicDirectory(
+ topic.id() as u32,
+ stream_id as u32,
+ topic_path,
+ ));
+ }
+ shard_info!(
+ shard_id,
+ "Saved topic with ID: {}. for stream with ID: {}",
+ topic.id(),
+ stream_id
+ );
+ /*
+ if !Path::new(&topic.partitions_path).exists()
+ && create_dir_all(&topic.partitions_path).await.is_err()
+ {
+ return Err(IggyError::CannotCreatePartitionsDirectory(
+ topic.stream_id,
+ topic.topic_id,
+ ));
+ }
+
+ shard_info!(shard_id,
+ "Saving {} partition(s) for topic {topic}...",
+ topic.partitions.len()
+ );
+ for (_, partition) in topic.partitions.iter() {
+ let mut partition = partition.write().await;
+ partition.persist().await.with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to persist partition, topic: {topic}"
+ )
+ })?;
+ }
+
+ shard_info!(shard_id, "Saved topic {topic}");
+ */
+ Ok(())
+}
diff --git a/core/server/src/streaming/topics/topic.rs b/core/server/src/streaming/topics/topic.rs
index 2884da05..b4fcba69 100644
--- a/core/server/src/streaming/topics/topic.rs
+++ b/core/server/src/streaming/topics/topic.rs
@@ -114,8 +114,8 @@ impl Topic {
max_topic_size: MaxTopicSize,
replication_factor: u8,
) -> Result<CreatedTopicInfo, IggyError> {
- let path = config.get_topic_path(stream_id, topic_id);
- let partitions_path = config.get_partitions_path(stream_id, topic_id);
+ let path = config.get_topic_path(stream_id as usize, topic_id as usize);
+ let partitions_path = config.get_partitions_path(stream_id as usize, topic_id as usize);
let mut topic = Topic {
stream_id,
topic_id,
@@ -316,8 +316,8 @@ mod tests {
let segments_count_of_parent_stream = Arc::new(AtomicU32::new(0));
let topic_info = Topic::create(
- stream_id,
- topic_id,
+ stream_id as u32,
+ topic_id as u32,
name,
partitions_count,
config,
@@ -332,8 +332,8 @@ mod tests {
)
.unwrap();
let topic = topic_info.topic;
- assert_eq!(topic.stream_id, stream_id);
- assert_eq!(topic.topic_id, topic_id);
+ assert_eq!(topic.stream_id, stream_id as u32);
+ assert_eq!(topic.topic_id, topic_id as u32);
assert_eq!(topic.path, path);
assert_eq!(topic.name, name);
assert_eq!(topic.partitions.len(), partitions_count as usize);
@@ -341,7 +341,7 @@ mod tests {
for (id, partition) in topic.partitions {
let partition = partition.read().await;
- assert_eq!(partition.stream_id, stream_id);
+ assert_eq!(partition.stream_id, stream_id as u32);
assert_eq!(partition.topic_id, topic.topic_id);
assert_eq!(partition.partition_id, id);
assert_eq!(partition.segments.len(), 1);
diff --git a/core/server/src/streaming/topics/topic2.rs b/core/server/src/streaming/topics/topic2.rs
index 70cc04d4..e2cc63d3 100644
--- a/core/server/src/streaming/topics/topic2.rs
+++ b/core/server/src/streaming/topics/topic2.rs
@@ -3,9 +3,6 @@ use iggy_common::{CompressionAlgorithm, IggyExpiry, IggyTimestamp, MaxTopicSize}
#[derive(Default, Debug)]
pub struct Topic {
- partitions: Partitions,
- consumer_groups: ConsumerGroups,
-
id: usize,
// TODO: This property should be removed, we won't use it in our clustering impl.
replication_factor: u8,
@@ -14,6 +11,11 @@ pub struct Topic {
message_expiry: IggyExpiry,
compression_algorithm: CompressionAlgorithm,
max_topic_size: MaxTopicSize,
+
+ partitions: Partitions,
+ consumer_groups: ConsumerGroups,
+ //consumer_offsets: (),
+ //consumer_group_offsets: (),
}
impl Topic {