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 11:40:37 UTC
(iggy) branch implement_consumer_and_cg_offsets created (now 51b920a4)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch implement_consumer_and_cg_offsets
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 51b920a4 feat(io_uring): implement consumer/cg offsets storage
This branch includes the following new commits:
new 51b920a4 feat(io_uring): implement consumer/cg offsets storage
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): implement consumer/cg offsets storage
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch implement_consumer_and_cg_offsets
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 51b920a4e5052fd34614d0565fccda07b3c26926
Author: numminex <g....@gmail.com>
AuthorDate: Wed Aug 6 13:40:25 2025 +0200
feat(io_uring): implement consumer/cg offsets storage
---
.../tests/streaming/common/test_setup.rs | 60 ++++++----
core/integration/tests/streaming/segment.rs | 32 ++++--
core/integration/tests/streaming/topic.rs | 16 ++-
.../binary/handlers/topics/create_topic_handler.rs | 14 ++-
core/server/src/shard/mod.rs | 2 -
core/server/src/shard/system/partitions.rs | 34 ++++--
core/server/src/shard/system/topics.rs | 29 ++---
core/server/src/shard/transmission/event.rs | 1 -
core/server/src/slab/streams.rs | 125 ++++++++++++---------
core/server/src/streaming/partitions/mod.rs | 1 +
core/server/src/streaming/partitions/partition.rs | 23 ++--
core/server/src/streaming/partitions/storage2.rs | 92 +++++++++++++++
core/server/src/streaming/segments/segment.rs | 7 +-
core/server/src/streaming/streams/stream2.rs | 12 ++
core/server/src/streaming/topics/storage2.rs | 22 ++--
core/server/src/streaming/topics/topic2.rs | 39 ++++++-
16 files changed, 358 insertions(+), 151 deletions(-)
diff --git a/core/integration/tests/streaming/common/test_setup.rs b/core/integration/tests/streaming/common/test_setup.rs
index bfa2a128..02cb95a1 100644
--- a/core/integration/tests/streaming/common/test_setup.rs
+++ b/core/integration/tests/streaming/common/test_setup.rs
@@ -85,25 +85,41 @@ 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 as usize, topic_id as usize))
- .await
- .is_err()
+ 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 as usize, topic_id as usize))
- .await
- .unwrap();
+ fs::create_dir(
+ &self
+ .config
+ .get_topic_path(stream_id as usize, topic_id as usize),
+ )
+ .await
+ .unwrap();
}
}
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 as usize, topic_id as usize))
- .await
- .is_err()
+ 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 as usize, topic_id as usize))
- .await
- .unwrap();
+ fs::create_dir(
+ &self
+ .config
+ .get_partitions_path(stream_id as usize, topic_id as usize),
+ )
+ .await
+ .unwrap();
}
}
@@ -114,19 +130,19 @@ impl TestSetup {
partition_id: u32,
) {
self.create_partitions_directory(stream_id, topic_id).await;
- if fs::metadata(
- &self
- .config
- .get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize),
- )
+ if fs::metadata(&self.config.get_partition_path(
+ stream_id as usize,
+ topic_id as usize,
+ partition_id as usize,
+ ))
.await
.is_err()
{
- fs::create_dir(
- &self
- .config
- .get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize),
- )
+ fs::create_dir(&self.config.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 3e3bc7cf..e2d012b1 100644
--- a/core/integration/tests/streaming/segment.rs
+++ b/core/integration/tests/streaming/segment.rs
@@ -66,9 +66,11 @@ async fn should_persist_segment() {
.await;
segment.open().await.unwrap();
assert_persisted_segment(
- &setup
- .config
- .get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize),
+ &setup.config.get_partition_path(
+ stream_id as usize,
+ topic_id as usize,
+ partition_id as usize,
+ ),
start_offset,
)
.await;
@@ -103,9 +105,11 @@ async fn should_load_existing_segment_from_disk() {
.await;
segment.open().await.unwrap();
assert_persisted_segment(
- &setup
- .config
- .get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize),
+ &setup.config.get_partition_path(
+ stream_id as usize,
+ topic_id as usize,
+ partition_id as usize,
+ ),
start_offset,
)
.await;
@@ -173,9 +177,11 @@ async fn should_persist_and_load_segment_with_messages() {
.await;
segment.open().await.unwrap();
assert_persisted_segment(
- &setup
- .config
- .get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize),
+ &setup.config.get_partition_path(
+ stream_id as usize,
+ topic_id as usize,
+ partition_id as usize,
+ ),
start_offset,
)
.await;
@@ -259,9 +265,11 @@ async fn given_at_least_one_not_expired_message_segment_should_not_be_expired()
.await;
segment.open().await.unwrap();
assert_persisted_segment(
- &setup
- .config
- .get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize),
+ &setup.config.get_partition_path(
+ stream_id as usize,
+ topic_id as usize,
+ partition_id as usize,
+ ),
start_offset,
)
.await;
diff --git a/core/integration/tests/streaming/topic.rs b/core/integration/tests/streaming/topic.rs
index cf5ab718..26a02f96 100644
--- a/core/integration/tests/streaming/topic.rs
+++ b/core/integration/tests/streaming/topic.rs
@@ -60,7 +60,9 @@ async fn should_persist_topics_with_partitions_directories_and_info_file() {
assert_persisted_topic(
&topic.path,
- &setup.config.get_partitions_path(stream_id as usize, topic_id as usize),
+ &setup
+ .config
+ .get_partitions_path(stream_id as usize, topic_id as usize),
3,
)
.await;
@@ -97,7 +99,9 @@ async fn should_load_existing_topic_from_disk() {
topic.persist().await.unwrap();
assert_persisted_topic(
&topic.path,
- &setup.config.get_partitions_path(stream_id as usize, topic_id as usize),
+ &setup
+ .config
+ .get_partitions_path(stream_id as usize, topic_id as usize),
partitions_count,
)
.await;
@@ -174,7 +178,9 @@ async fn should_delete_existing_topic_from_disk() {
topic.persist().await.unwrap();
assert_persisted_topic(
&topic.path,
- &setup.config.get_partitions_path(stream_id as usize, topic_id as usize),
+ &setup
+ .config
+ .get_partitions_path(stream_id as usize, topic_id as usize),
partitions_count,
)
.await;
@@ -214,7 +220,9 @@ async fn should_purge_existing_topic_on_disk() {
topic.persist().await.unwrap();
assert_persisted_topic(
&topic.path,
- &setup.config.get_partitions_path(stream_id as usize, topic_id as usize),
+ &setup
+ .config
+ .get_partitions_path(stream_id as usize, topic_id as usize),
partitions_count,
)
.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 db7d409e..bb626e5d 100644
--- a/core/server/src/binary/handlers/topics/create_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/create_topic_handler.rs
@@ -32,9 +32,9 @@ use crate::streaming::stats::stats::TopicStats;
use crate::streaming::topics::topic::Topic;
use anyhow::Result;
use error_set::ErrContext;
-use iggy_common::IggyError;
use iggy_common::create_topic::CreateTopic;
use iggy_common::locking::IggyRwLockFn;
+use iggy_common::{Identifier, IggyError};
use std::rc::Rc;
use std::sync::Arc;
use tracing::{debug, instrument};
@@ -65,7 +65,6 @@ impl ServerCommandHandler for CreateTopic {
&stream_id,
maybe_topic_id,
self.name.clone(),
- self.partitions_count,
self.message_expiry,
self.compression_algorithm,
self.max_topic_size,
@@ -73,11 +72,20 @@ impl ServerCommandHandler for CreateTopic {
stats.clone(),
)
.await?;
+
+ shard
+ .create_partitions2(
+ session,
+ &stream_id,
+ &Identifier::numeric(new_topic_id as u32).unwrap(),
+ self.partitions_count,
+ )
+ .await?;
+
let event = ShardEvent::CreatedTopic2 {
stream_id: self.stream_id.clone(),
id: new_topic_id,
name: self.name.clone(),
- partitions_count: self.partitions_count,
message_expiry: self.message_expiry,
compression_algorithm: self.compression_algorithm,
max_topic_size: self.max_topic_size,
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 82bd3e33..52679f63 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -837,7 +837,6 @@ impl IggyShard {
id,
stream_id,
name,
- partitions_count,
message_expiry,
compression_algorithm,
max_topic_size,
@@ -847,7 +846,6 @@ impl IggyShard {
let topic_id = self.create_topic2_bypass_auth(
stream_id,
name.to_owned(),
- *partitions_count,
*replication_factor,
*message_expiry,
*compression_algorithm,
diff --git a/core/server/src/shard/system/partitions.rs b/core/server/src/shard/system/partitions.rs
index 487fc0f7..f06d72bc 100644
--- a/core/server/src/shard/system/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -19,7 +19,9 @@
use super::COMPONENT;
use crate::shard::IggyShard;
use crate::streaming::partitions::partition2;
+use crate::streaming::partitions::storage2::create_partition_file_hierarchy;
use crate::streaming::session::Session;
+use ahash::AHashMap;
use error_set::ErrContext;
use iggy_common::Identifier;
use iggy_common::IggyError;
@@ -39,6 +41,8 @@ impl IggyShard {
// Create partitions...
let partition_ids = std::iter::repeat_with(partition2::Partition::new)
.map(|partition| {
+ topic.consumer_group_offsets_mut().insert(AHashMap::new());
+ topic.consumer_offsets_mut().insert(AHashMap::new());
topic
.partitions_mut()
.with_mut(|partitions| partition.insert_into(partitions))
@@ -58,13 +62,13 @@ impl IggyShard {
partitions_count: u32,
) -> Result<(), IggyError> {
self.ensure_authenticated(session)?;
+ let numeric_stream_id = self
+ .streams2
+ .with_stream_by_id(stream_id, |stream| stream.id());
+ let numeric_topic_id = self
+ .streams2
+ .with_topic_by_id(stream_id, topic_id, |topic| topic.id());
{
- let numeric_stream_id = self
- .streams2
- .with_stream_by_id(stream_id, |stream| stream.id());
- let numeric_topic_id = self
- .streams2
- .with_topic_by_id(stream_id, topic_id, |topic| topic.id());
self.permissioner.borrow().create_partitions(
session.get_user_id(),
numeric_stream_id as u32,
@@ -76,7 +80,23 @@ impl IggyShard {
numeric_topic_id
))?;
}
- self.create_partitions2_base(stream_id, topic_id, partitions_count);
+
+ let partition_ids = self.create_partitions2_base(stream_id, topic_id, partitions_count);
+ // Create file hierarchy for partition
+ for partition_id in partition_ids {
+ create_partition_file_hierarchy(
+ self.id,
+ numeric_stream_id,
+ numeric_topic_id,
+ partition_id,
+ &self.config.system,
+ )
+ .await?;
+
+ // And open descriptors for partitions that belong to _this_ shard.
+ // Or maybe not, maybe those should be opened lazily ?
+ // I think the lazy option is better - Claude Sonnet 4
+ }
Ok(())
}
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index 46c6b5d2..e7573ed3 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -25,9 +25,11 @@ use crate::shard::transmission::event::ShardEvent;
use crate::shard::{IggyShard, ShardInfo};
use crate::shard_info;
use crate::streaming::partitions::partition2;
+use crate::streaming::partitions::storage2::create_partition_file_hierarchy;
use crate::streaming::session::Session;
use crate::streaming::stats::stats::TopicStats;
use crate::streaming::streams::stream::Stream;
+use crate::streaming::topics::storage2::create_topic_file_hierarchy;
use crate::streaming::topics::topic::Topic;
use crate::streaming::topics::topic2;
use clap::Id;
@@ -146,7 +148,6 @@ impl IggyShard {
stream_id: &Identifier,
topic_id: Option<u32>,
name: String,
- partitions_count: u32,
message_expiry: IggyExpiry,
compression: CompressionAlgorithm,
max_topic_size: MaxTopicSize,
@@ -155,13 +156,13 @@ impl IggyShard {
) -> Result<usize, IggyError> {
self.ensure_authenticated(session)?;
//self.ensure_stream_exists(stream_id)?;
+ let numeric_stream_id = self
+ .streams2
+ .with_stream_by_id(stream_id, |stream| stream.id());
{
- let stream_id = self
- .streams2
- .with_stream_by_id(stream_id, |stream| stream.id());
self.permissioner
.borrow()
- .create_topic(session.get_user_id(), stream_id as u32)
+ .create_topic(session.get_user_id(), numeric_stream_id as u32)
.with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - permission denied to create topic with name: {name} in stream with ID: {stream_id} for user with ID: {}",
@@ -191,14 +192,9 @@ 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(),
- partitions_count,
- );
- // TODO: Create partition files and open descriptors for partitions that belong to _this_ shard.
+ // Create file hierarchy for the topic.
+ create_topic_file_hierarchy(self.id, numeric_stream_id, topic_id, &self.config.system)
+ .await?;
Ok(topic_id)
}
@@ -206,7 +202,6 @@ impl IggyShard {
&self,
stream_id: &Identifier,
name: String,
- partitions_count: u32,
replication_factor: Option<u8>,
message_expiry: IggyExpiry,
compression: CompressionAlgorithm,
@@ -222,12 +217,6 @@ impl IggyShard {
max_topic_size,
stats,
)?;
- let partition_ids = self.create_partitions2_base(
- stream_id,
- &Identifier::numeric(topic_id as u32).unwrap(),
- partitions_count,
- );
- // TODO: Open descriptors for partitions that belong to _this_ shard.
Ok(topic_id)
}
diff --git a/core/server/src/shard/transmission/event.rs b/core/server/src/shard/transmission/event.rs
index b967daf2..c5b3dfd8 100644
--- a/core/server/src/shard/transmission/event.rs
+++ b/core/server/src/shard/transmission/event.rs
@@ -75,7 +75,6 @@ pub enum ShardEvent {
id: usize,
stream_id: Identifier,
name: String,
- partitions_count: u32,
message_expiry: IggyExpiry,
compression_algorithm: CompressionAlgorithm,
max_topic_size: MaxTopicSize,
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index 448ec602..67112e12 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -82,35 +82,24 @@ impl Streams {
id: &Identifier,
f: impl FnOnce(&stream2::Stream) -> T,
) -> T {
- self.with(|streams| {
- let stream = match id.kind {
- iggy_common::IdKind::Numeric => {
- let id = id.get_u32_value().unwrap() as usize;
- &streams[id]
- }
- iggy_common::IdKind::String => {
- let key = id.get_string_value().unwrap();
- unsafe { streams.get_by_key_unchecked(&key) }
- }
- };
- f(stream)
- })
+ self.with(|streams| Self::get_stream_ref(id, streams).invoke(f))
}
- pub fn with_stream_by_id_mut(&self, id: &Identifier, f: impl FnOnce(&mut stream2::Stream)) {
- self.with_mut(|streams| {
- let stream = match id.kind {
- iggy_common::IdKind::Numeric => {
- let id = id.get_u32_value().unwrap() as usize;
- &mut streams[id]
- }
- iggy_common::IdKind::String => {
- let key = id.get_string_value().unwrap();
- unsafe { streams.get_by_key_mut_unchecked(&key) }
- }
- };
- f(stream)
- });
+ pub async fn with_stream_by_id_async<T>(
+ &self,
+ id: &Identifier,
+ f: impl AsyncFnOnce(&stream2::Stream) -> T,
+ ) -> T {
+ self.with_async(async |streams| Self::get_stream_ref(id, streams).invoke_async(f).await)
+ .await
+ }
+
+ pub fn with_stream_by_id_mut<T>(
+ &self,
+ id: &Identifier,
+ f: impl FnOnce(&mut stream2::Stream) -> T,
+ ) -> T {
+ self.with_mut(|streams| Self::get_stream_mut(id, streams).invoke_mut(f))
}
pub fn with_topics<T>(&self, stream_id: &Identifier, f: impl FnOnce(&Topics) -> T) -> T {
@@ -122,20 +111,8 @@ impl Streams {
stream_id: &Identifier,
f: impl AsyncFnOnce(&Topics) -> T,
) -> T {
- self.with_async(async |streams: &IndexedSlab<stream2::Stream>| {
- let stream = match stream_id.kind {
- iggy_common::IdKind::Numeric => {
- let id = stream_id.get_u32_value().unwrap() as usize;
- &streams[id]
- }
- iggy_common::IdKind::String => {
- let key = stream_id.get_string_value().unwrap();
- unsafe { streams.get_by_key_unchecked(&key) }
- }
- };
- f(stream.topics()).await
- })
- .await
+ self.with_stream_by_id_async(stream_id, async |stream| f(stream.topics()).await)
+ .await
}
pub fn with_topics_mut<T>(
@@ -143,19 +120,7 @@ impl Streams {
stream_id: &Identifier,
f: impl FnOnce(&mut Topics) -> T,
) -> T {
- self.with_mut(|streams| {
- let stream = match stream_id.kind {
- iggy_common::IdKind::Numeric => {
- let id = stream_id.get_u32_value().unwrap() as usize;
- &mut streams[id]
- }
- iggy_common::IdKind::String => {
- let key = stream_id.get_string_value().unwrap();
- unsafe { streams.get_by_key_mut_unchecked(&key) }
- }
- };
- f(stream.topics_mut())
- })
+ self.with_stream_by_id_mut(stream_id, |stream| f(stream.topics_mut()))
}
pub fn with_topic_by_id<T>(
@@ -167,13 +132,25 @@ impl Streams {
self.with_topics(stream_id, |topics| topics.with_topic_by_id(id, f))
}
- pub fn with_topic_by_id_mut<T>(
+ pub async fn with_topic_by_id_async<T>(
&self,
+ stream_id: &Identifier,
id: &Identifier,
+ f: impl AsyncFnOnce(&topic2::Topic) -> T,
+ ) -> T {
+ self.with_topics_async(stream_id, async |topics| {
+ topics.with_topic_by_id_async(id, f).await
+ })
+ .await
+ }
+
+ pub fn with_topic_by_id_mut<T>(
+ &self,
+ stream_id: &Identifier,
topic_id: &Identifier,
f: impl FnOnce(&mut topic2::Topic) -> T,
) -> T {
- self.with_topics(id, |topics| topics.with_topic_by_id_mut(topic_id, f))
+ self.with_topics_mut(stream_id, |topics| topics.with_topic_by_id_mut(topic_id, f))
}
pub fn with_partitions(
@@ -210,4 +187,40 @@ impl Streams {
partitions.with_partition_id(partition_id, f);
});
}
+
+ fn get_stream_ref<'streams>(
+ id: &Identifier,
+ slab: &'streams IndexedSlab<stream2::Stream>,
+ ) -> &'streams stream2::Stream {
+ match id.kind {
+ iggy_common::IdKind::Numeric => {
+ let idx = id.get_u32_value().unwrap() as usize;
+ &slab[idx]
+ }
+ iggy_common::IdKind::String => {
+ let key = id.get_string_value().unwrap();
+ unsafe { slab.get_by_key_unchecked(&key) }
+ }
+ }
+ }
+
+ fn get_stream_mut<'streams>(
+ id: &Identifier,
+ slab: &'streams mut IndexedSlab<stream2::Stream>,
+ ) -> &'streams mut stream2::Stream {
+ match id.kind {
+ iggy_common::IdKind::Numeric => {
+ let idx = id.get_u32_value().unwrap() as usize;
+ &mut slab[idx]
+ }
+ iggy_common::IdKind::String => {
+ let key = id.get_string_value().unwrap();
+ unsafe { slab.get_by_key_mut_unchecked(&key) }
+ }
+ }
+ }
+
+ pub fn len(&self) -> usize {
+ self.container.borrow().len()
+ }
}
diff --git a/core/server/src/streaming/partitions/mod.rs b/core/server/src/streaming/partitions/mod.rs
index fb9e936a..e9d62716 100644
--- a/core/server/src/streaming/partitions/mod.rs
+++ b/core/server/src/streaming/partitions/mod.rs
@@ -23,5 +23,6 @@ pub mod partition2;
pub mod persistence;
pub mod segments;
pub mod storage;
+pub mod storage2;
pub const COMPONENT: &str = "STREAMING_PARTITIONS";
diff --git a/core/server/src/streaming/partitions/partition.rs b/core/server/src/streaming/partitions/partition.rs
index 3103e2af..83790946 100644
--- a/core/server/src/streaming/partitions/partition.rs
+++ b/core/server/src/streaming/partitions/partition.rs
@@ -103,12 +103,20 @@ impl Partition {
segments_count_of_parent_stream: Arc<AtomicU32>,
created_at: IggyTimestamp,
) -> Partition {
- 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 as usize, topic_id as usize, partition_id as usize);
- let consumer_group_offsets_path =
- config.get_consumer_group_offsets_path(stream_id as usize, topic_id as usize, partition_id as usize);
+ 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 as usize,
+ topic_id as usize,
+ partition_id as usize,
+ );
+ let consumer_group_offsets_path = 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(
@@ -244,7 +252,8 @@ mod tests {
let topic_id = 2usize;
let partition_id = 3usize;
let with_segment = true;
- let path = config.get_partition_path(stream_id as usize, topic_id as usize, partition_id as usize);
+ 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 as u32,
diff --git a/core/server/src/streaming/partitions/storage2.rs b/core/server/src/streaming/partitions/storage2.rs
new file mode 100644
index 00000000..d733fbf4
--- /dev/null
+++ b/core/server/src/streaming/partitions/storage2.rs
@@ -0,0 +1,92 @@
+use crate::{configs::system::SystemConfig, shard_error, shard_info};
+use compio::fs::create_dir_all;
+use iggy_common::IggyError;
+use std::path::Path;
+
+pub async fn create_partition_file_hierarchy(
+ shard_id: u16,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ config: &SystemConfig,
+) -> Result<(), IggyError> {
+ let partition_path = config.get_partition_path(stream_id, topic_id, partition_id);
+ shard_info!(
+ shard_id,
+ "Saving partition with ID: {} for stream with ID: {} and topic with ID: {}...",
+ partition_id,
+ stream_id,
+ topic_id
+ );
+ if !Path::new(&partition_path).exists() && create_dir_all(&partition_path).await.is_err() {
+ return Err(IggyError::CannotCreatePartitionDirectory(
+ partition_id as u32,
+ stream_id as u32,
+ topic_id as u32,
+ ));
+ }
+
+ let offset_path = config.get_offsets_path(stream_id, topic_id, partition_id);
+ if !Path::new(&offset_path).exists() && create_dir_all(&offset_path).await.is_err() {
+ shard_error!(
+ shard_id,
+ "Failed to create offsets directory for partition with ID: {} for stream with ID: {} and topic with ID: {}.",
+ partition_id,
+ stream_id,
+ topic_id
+ );
+ return Err(IggyError::CannotCreatePartition(
+ partition_id as u32,
+ stream_id as u32,
+ topic_id as u32,
+ ));
+ }
+
+ let consumer_offset_path = config.get_consumer_offsets_path(stream_id, topic_id, partition_id);
+ if !Path::new(&consumer_offset_path).exists()
+ && create_dir_all(&consumer_offset_path).await.is_err()
+ {
+ shard_error!(
+ shard_id,
+ "Failed to create consumer offsets directory for partition with ID: {} for stream with ID: {} and topic with ID: {}.",
+ partition_id,
+ stream_id,
+ topic_id
+ );
+ return Err(IggyError::CannotCreatePartition(
+ partition_id as u32,
+ stream_id as u32,
+ topic_id as u32,
+ ));
+ }
+
+ let consumer_group_offsets_path =
+ config.get_consumer_group_offsets_path(stream_id, topic_id, partition_id);
+ if !Path::new(&consumer_group_offsets_path).exists()
+ && create_dir_all(&consumer_group_offsets_path).await.is_err()
+ {
+ shard_error!(
+ shard_id,
+ "Failed to create consumer group offsets directory for partition with ID: {} for stream with ID: {} and topic with ID: {}.",
+ partition_id,
+ stream_id,
+ topic_id
+ );
+ return Err(IggyError::CannotCreatePartition(
+ partition_id as u32,
+ stream_id as u32,
+ topic_id as u32,
+ ));
+ }
+
+ shard_info!(
+ shard_id,
+ "Saved partition with start ID: {} for stream with ID: {} and topic with ID: {}, path: {}.",
+ partition_id,
+ stream_id,
+ topic_id,
+ partition_path
+ );
+
+ Ok(())
+}
diff --git a/core/server/src/streaming/segments/segment.rs b/core/server/src/streaming/segments/segment.rs
index b197cb41..5da1018f 100644
--- a/core/server/src/streaming/segments/segment.rs
+++ b/core/server/src/streaming/segments/segment.rs
@@ -85,7 +85,12 @@ 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 as usize, topic_id as usize, partition_id as usize, 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 {
diff --git a/core/server/src/streaming/streams/stream2.rs b/core/server/src/streaming/streams/stream2.rs
index 88b4e64b..2dc6743e 100644
--- a/core/server/src/streaming/streams/stream2.rs
+++ b/core/server/src/streaming/streams/stream2.rs
@@ -41,6 +41,18 @@ impl Stream {
}
}
+ pub fn invoke<T>(&self, f: impl FnOnce(&Self) -> T) -> T {
+ f(self)
+ }
+
+ pub fn invoke_mut<T>(&mut self, f: impl FnOnce(&mut Self) -> T) -> T {
+ f(self)
+ }
+
+ pub async fn invoke_async<T>(&self, f: impl AsyncFnOnce(&Self) -> T) -> T {
+ f(self).await
+ }
+
pub fn id(&self) -> usize {
self.id
}
diff --git a/core/server/src/streaming/topics/storage2.rs b/core/server/src/streaming/topics/storage2.rs
index ccc8f164..1fc2b58f 100644
--- a/core/server/src/streaming/topics/storage2.rs
+++ b/core/server/src/streaming/topics/storage2.rs
@@ -5,16 +5,17 @@ use std::path::Path;
use crate::{configs::system::SystemConfig, shard_info, streaming::topics::topic2};
-async fn create_topic_file_hierarchy(
+pub async fn create_topic_file_hierarchy(
shard_id: u16,
stream_id: usize,
- topic: &topic2::Topic,
+ topic_id: usize,
config: &SystemConfig,
) -> Result<(), IggyError> {
- let topic_path = config.get_topic_path(stream_id, topic.id());
+ let topic_path = config.get_topic_path(stream_id, topic_id);
+ let partitions_path = config.get_partitions_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,
+ topic_id as u32,
stream_id as u32,
topic_path,
));
@@ -22,19 +23,18 @@ async fn create_topic_file_hierarchy(
shard_info!(
shard_id,
"Saved topic with ID: {}. for stream with ID: {}",
- topic.id(),
+ topic_id,
stream_id
);
- /*
- if !Path::new(&topic.partitions_path).exists()
- && create_dir_all(&topic.partitions_path).await.is_err()
- {
+
+ if !Path::new(&partitions_path).exists() && create_dir_all(&partitions_path).await.is_err() {
return Err(IggyError::CannotCreatePartitionsDirectory(
- topic.stream_id,
- topic.topic_id,
+ stream_id as u32,
+ topic_id as u32,
));
}
+ /*
shard_info!(shard_id,
"Saving {} partition(s) for topic {topic}...",
topic.partitions.len()
diff --git a/core/server/src/streaming/topics/topic2.rs b/core/server/src/streaming/topics/topic2.rs
index e2cc63d3..d36b7475 100644
--- a/core/server/src/streaming/topics/topic2.rs
+++ b/core/server/src/streaming/topics/topic2.rs
@@ -1,5 +1,15 @@
-use crate::slab::{IndexedSlab, Keyed, consumer_groups::ConsumerGroups, partitions::Partitions};
+use crate::{
+ slab::{
+ IndexedSlab, Keyed,
+ consumer_groups::ConsumerGroups,
+ partitions::{PARTITIONS_CAPACITY, Partitions},
+ },
+ streaming::partitions::partition::ConsumerOffset,
+};
+use ahash::AHashMap;
+use dashmap::DashMap;
use iggy_common::{CompressionAlgorithm, IggyExpiry, IggyTimestamp, MaxTopicSize};
+use slab::Slab;
#[derive(Default, Debug)]
pub struct Topic {
@@ -14,8 +24,9 @@ pub struct Topic {
partitions: Partitions,
consumer_groups: ConsumerGroups,
- //consumer_offsets: (),
- //consumer_group_offsets: (),
+
+ consumer_offsets: Slab<AHashMap<usize, ConsumerOffset>>,
+ consumer_group_offsets: Slab<AHashMap<usize, ConsumerOffset>>,
}
impl Topic {
@@ -30,12 +41,14 @@ impl Topic {
id: 0,
name,
created_at: IggyTimestamp::now(),
- partitions: Partitions::default(),
- consumer_groups: ConsumerGroups::default(),
replication_factor,
message_expiry,
compression_algorithm: compression,
max_topic_size,
+ partitions: Partitions::default(),
+ consumer_groups: ConsumerGroups::default(),
+ consumer_offsets: Slab::with_capacity(PARTITIONS_CAPACITY),
+ consumer_group_offsets: Slab::with_capacity(PARTITIONS_CAPACITY),
}
}
@@ -99,6 +112,22 @@ impl Topic {
&mut self.consumer_groups
}
+ pub fn consumer_offsets(&self) -> &Slab<AHashMap<usize, ConsumerOffset>> {
+ &self.consumer_offsets
+ }
+
+ pub fn consumer_offsets_mut(&mut self) -> &mut Slab<AHashMap<usize, ConsumerOffset>> {
+ &mut self.consumer_offsets
+ }
+
+ pub fn consumer_group_offsets(&self) -> &Slab<AHashMap<usize, ConsumerOffset>> {
+ &self.consumer_group_offsets
+ }
+
+ pub fn consumer_group_offsets_mut(&mut self) -> &mut Slab<AHashMap<usize, ConsumerOffset>> {
+ &mut self.consumer_group_offsets
+ }
+
pub fn insert_into(self, container: &mut IndexedSlab<Self>) -> usize {
let idx = container.insert(self);
let topic = &mut container[idx];