You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/09/15 20:27:46 UTC
(iggy) branch io_uring_tpc updated: feat(io_uring): fix integration tests for messages (#2170)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk 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 a93e5204 feat(io_uring): fix integration tests for messages (#2170)
a93e5204 is described below
commit a93e52042f03e7b8f3db4cffb2b396d221cd5979
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Mon Sep 15 22:27:43 2025 +0200
feat(io_uring): fix integration tests for messages (#2170)
---
core/common/src/error/iggy_error.rs | 3 +
core/integration/tests/archiver/disk.rs | 132 -----
core/integration/tests/archiver/mod.rs | 58 ---
core/integration/tests/mod.rs | 1 -
.../integration/tests/streaming/consumer_offset.rs | 83 ---
core/integration/tests/streaming/get_by_offset.rs | 139 +++--
.../tests/streaming/get_by_timestamp.rs | 74 +--
core/integration/tests/streaming/messages.rs | 164 ------
core/integration/tests/streaming/mod.rs | 110 +++-
core/integration/tests/streaming/partition.rs | 225 --------
core/integration/tests/streaming/segment.rs | 575 ---------------------
core/integration/tests/streaming/shard.rs | 170 ------
core/integration/tests/streaming/stream.rs | 191 -------
core/integration/tests/streaming/topic.rs | 279 ----------
core/integration/tests/streaming/topic_messages.rs | 245 ---------
.../handlers/messages/poll_messages_handler.rs | 4 +-
.../handlers/messages/send_messages_handler.rs | 4 +-
core/server/src/configs/system.rs | 27 +
core/server/src/http/http_shard_wrapper.rs | 8 +-
core/server/src/http/messages.rs | 8 +-
core/server/src/shard/mod.rs | 79 +--
core/server/src/shard/namespace.rs | 35 ++
core/server/src/shard/system/messages.rs | 284 ++--------
core/server/src/shard/system/partitions.rs | 61 +--
core/server/src/shard/system/streams.rs | 11 +-
core/server/src/shard/system/topics.rs | 9 +-
core/server/src/slab/streams.rs | 257 ++++++++-
core/server/src/streaming/local_sizeable.rs | 31 --
core/server/src/streaming/mod.rs | 2 +-
core/server/src/streaming/partitions/helpers.rs | 32 +-
core/server/src/streaming/partitions/journal.rs | 2 +-
core/server/src/streaming/partitions/partition2.rs | 54 +-
.../streaming/segments/types/messages_batch_set.rs | 8 -
core/server/src/streaming/streams/messages.rs | 111 ----
core/server/src/streaming/streams/mod.rs | 1 -
core/server/src/streaming/streams/stream2.rs | 15 +-
core/server/src/streaming/topics/messages.rs | 362 -------------
core/server/src/streaming/topics/mod.rs | 1 -
core/server/src/streaming/topics/storage2.rs | 17 -
core/server/src/streaming/topics/topic2.rs | 55 +-
core/server/src/streaming/traits.rs | 25 +
41 files changed, 742 insertions(+), 3210 deletions(-)
diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index c1ac5222..614757f1 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -466,6 +466,9 @@ pub enum IggyError {
#[error("Cannot read index timestamp")]
CannotReadIndexTimestamp = 10012,
+ #[error("Timestamp out of range: {0}")]
+ TimestampOutOfRange(u64) = 10013,
+
#[error("Shard not found for stream ID: {0}, topic ID: {1}, partition ID: {2}")]
ShardNotFound(usize, usize, usize) = 11000,
#[error("Shard communication error, shard ID: {0}")]
diff --git a/core/integration/tests/archiver/disk.rs b/core/integration/tests/archiver/disk.rs
deleted file mode 100644
index d99ce8ef..00000000
--- a/core/integration/tests/archiver/disk.rs
+++ /dev/null
@@ -1,132 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::archiver::DiskArchiverSetup;
-use compio::io::{AsyncReadAtExt, AsyncWriteAtExt};
-use server::streaming::utils::file;
-use server::{archiver::Archiver, server_error::ArchiverError};
-use std::path::Path;
-
-#[compio::test]
-async fn should_init_base_archiver_directory() {
- let setup = DiskArchiverSetup::init().await;
- let archiver = setup.archiver();
- let result = archiver.init().await;
- assert!(result.is_ok());
- let path = Path::new(&setup.archive_path);
- assert!(path.exists());
-}
-
-#[compio::test]
-async fn should_archive_file_on_disk_by_making_a_copy_of_original_file() {
- let setup = DiskArchiverSetup::init().await;
- let archiver = setup.archiver();
- let content = "hello world";
- let file_to_archive_path = format!("{}/file_to_archive", setup.base_path);
- create_file(&file_to_archive_path, content).await;
- let files_to_archive = vec![file_to_archive_path.as_ref()];
-
- let result = archiver.archive(&files_to_archive, None).await;
- assert!(result.is_ok());
- let archived_file_path = format!("{}/{}", setup.archive_path, file_to_archive_path);
- assert_archived_file(&file_to_archive_path, &archived_file_path, content).await;
-}
-
-#[compio::test]
-async fn should_archive_file_on_disk_within_additional_base_directory() {
- let setup = DiskArchiverSetup::init().await;
- let archiver = setup.archiver();
- let base_directory = "base";
- let content = "hello world";
- let file_to_archive_path = format!("{}/file_to_archive", setup.base_path);
- create_file(&file_to_archive_path, content).await;
- let files_to_archive = vec![file_to_archive_path.as_ref()];
-
- let result = archiver
- .archive(&files_to_archive, Some(base_directory.to_string()))
- .await;
- assert!(result.is_ok());
- let archived_file_path = format!(
- "{}/{base_directory}/{}",
- setup.archive_path, file_to_archive_path
- );
- assert_archived_file(&file_to_archive_path, &archived_file_path, content).await;
-}
-
-#[compio::test]
-async fn should_return_true_when_file_is_archived() {
- let setup = DiskArchiverSetup::init().await;
- let archiver = setup.archiver();
- let content = "hello world";
- let file_to_archive_path = format!("{}/file_to_archive", setup.base_path);
- create_file(&file_to_archive_path, content).await;
- let files_to_archive = vec![file_to_archive_path.as_ref()];
- archiver.archive(&files_to_archive, None).await.unwrap();
-
- let is_archived = archiver.is_archived(&file_to_archive_path, None).await;
- assert!(is_archived.is_ok());
- assert!(is_archived.unwrap());
-}
-
-#[compio::test]
-async fn should_return_false_when_file_is_not_archived() {
- let setup = DiskArchiverSetup::init().await;
- let archiver = setup.archiver();
- let content = "hello world";
- let file_to_archive_path = format!("{}/file_to_archive", setup.base_path);
- create_file(&file_to_archive_path, content).await;
-
- let is_archived = archiver.is_archived(&file_to_archive_path, None).await;
- assert!(is_archived.is_ok());
- assert!(!is_archived.unwrap());
-}
-
-#[compio::test]
-async fn should_fail_when_file_to_archive_does_not_exist() {
- let setup = DiskArchiverSetup::init().await;
- let archiver = setup.archiver();
- let file_to_archive_path = "invalid_file_to_archive";
- let files_to_archive = vec![file_to_archive_path];
- let result = archiver.archive(&files_to_archive, None).await;
-
- assert!(result.is_err());
- let error = result.err().unwrap();
- assert!(matches!(error, ArchiverError::FileToArchiveNotFound { .. }));
-}
-
-async fn create_file(path: &str, content: &str) {
- let mut file = file::overwrite(path).await.unwrap();
- let content = content.as_bytes().to_vec();
- file.write_all_at(content, 0).await.unwrap();
-}
-
-async fn assert_archived_file(file_to_archive_path: &str, archived_file_path: &str, content: &str) {
- assert!(Path::new(&file_to_archive_path).exists());
- assert!(Path::new(&archived_file_path).exists());
- let archived_file = file::open(archived_file_path).await;
- assert!(archived_file.is_ok());
- let archived_file = archived_file.unwrap();
- let len = archived_file.metadata().await.unwrap().len();
- let archived_file_content = Vec::with_capacity(len as usize);
- let (_, archived_file_content) = archived_file
- .read_exact_at(archived_file_content, 0)
- .await
- .unwrap();
- let archived_file_content = String::from_utf8(archived_file_content).unwrap();
- assert_eq!(content, archived_file_content);
-}
diff --git a/core/integration/tests/archiver/mod.rs b/core/integration/tests/archiver/mod.rs
deleted file mode 100644
index 704936e9..00000000
--- a/core/integration/tests/archiver/mod.rs
+++ /dev/null
@@ -1,58 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use compio::fs::create_dir;
-use server::archiver::disk::DiskArchiver;
-use server::configs::server::DiskArchiverConfig;
-use uuid::Uuid;
-
-mod disk;
-
-pub struct DiskArchiverSetup {
- base_path: String,
- archive_path: String,
- archiver: DiskArchiver,
-}
-
-impl DiskArchiverSetup {
- pub async fn init() -> DiskArchiverSetup {
- let base_path = format!("test_local_data_{}", Uuid::now_v7().to_u128_le());
- let archive_path = format!("{base_path}/archive");
- let config = DiskArchiverConfig {
- path: archive_path.clone(),
- };
- let archiver = DiskArchiver::new(config);
- create_dir(&base_path).await.unwrap();
-
- Self {
- base_path,
- archive_path,
- archiver,
- }
- }
-
- pub fn archiver(&self) -> &DiskArchiver {
- &self.archiver
- }
-}
-
-impl Drop for DiskArchiverSetup {
- fn drop(&mut self) {
- std::fs::remove_dir_all(&self.base_path).unwrap();
- }
-}
diff --git a/core/integration/tests/mod.rs b/core/integration/tests/mod.rs
index ddf74aa4..2adf4fd0 100644
--- a/core/integration/tests/mod.rs
+++ b/core/integration/tests/mod.rs
@@ -25,7 +25,6 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Once};
use std::{panic, thread};
-mod archiver;
mod cli;
mod config_provider;
mod data_integrity;
diff --git a/core/integration/tests/streaming/consumer_offset.rs b/core/integration/tests/streaming/consumer_offset.rs
deleted file mode 100644
index db89169b..00000000
--- a/core/integration/tests/streaming/consumer_offset.rs
+++ /dev/null
@@ -1,83 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::common::test_setup::TestSetup;
-use compio::fs;
-use iggy::prelude::ConsumerKind;
-use server::configs::system::SystemConfig;
-use server::streaming::partitions::partition::ConsumerOffset;
-use server::streaming::storage::PartitionStorageKind;
-use std::sync::Arc;
-
-#[compio::test]
-async fn should_persist_consumer_offsets_and_then_load_them_from_disk() {
- let setup = TestSetup::init().await;
- let storage = setup.storage.partition.as_ref();
- assert_persisted_offsets(&setup.config, storage, ConsumerKind::Consumer).await;
- assert_persisted_offsets(&setup.config, storage, ConsumerKind::ConsumerGroup).await;
-}
-
-async fn assert_persisted_offsets(
- config: &Arc<SystemConfig>,
- storage: &PartitionStorageKind,
- kind: ConsumerKind,
-) {
- let consumer_ids_count = 3;
- let offsets_count = 5;
- let path = match kind {
- ConsumerKind::Consumer => "consumer_offsets",
- ConsumerKind::ConsumerGroup => "consumer_group_offsets",
- };
- let path = format!("{}/{}", config.get_system_path(), path);
- fs::create_dir(&path).await.unwrap();
- for consumer_id in 1..=consumer_ids_count {
- let expected_offsets_count = consumer_id;
- for offset in 0..=offsets_count {
- let consumer_offset = ConsumerOffset::new(kind, consumer_id, offset, &path);
- assert_persisted_offset(&path, storage, &consumer_offset, expected_offsets_count).await;
- }
- }
-}
-
-async fn assert_persisted_offset(
- path: &str,
- storage: &PartitionStorageKind,
- consumer_offset: &ConsumerOffset,
- expected_offsets_count: u32,
-) {
- storage
- .save_consumer_offset(consumer_offset.offset, &consumer_offset.path)
- .await
- .unwrap();
- let consumer_offsets = storage
- .load_consumer_offsets(consumer_offset.kind, path)
- .await
- .unwrap();
- let expected_offsets_count = expected_offsets_count as usize;
- assert_eq!(consumer_offsets.len(), expected_offsets_count);
- let loaded_consumer_offset = consumer_offsets.get(expected_offsets_count - 1).unwrap();
-
- assert_eq!(loaded_consumer_offset.offset, consumer_offset.offset);
-
- assert_eq!(loaded_consumer_offset.kind, consumer_offset.kind);
- assert_eq!(
- loaded_consumer_offset.consumer_id,
- consumer_offset.consumer_id
- );
- assert_eq!(loaded_consumer_offset.path, consumer_offset.path);
-}
diff --git a/core/integration/tests/streaming/get_by_offset.rs b/core/integration/tests/streaming/get_by_offset.rs
index 7f104858..8044284a 100644
--- a/core/integration/tests/streaming/get_by_offset.rs
+++ b/core/integration/tests/streaming/get_by_offset.rs
@@ -1,32 +1,37 @@
/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses let consumer = PollingConsumer::consumer(&Identifier::numeric(2).unwrap(), partition_id as usize);
+ let args = PollingArgs::new(PollingStrategy::offset(middle_offset + 1), remaining_messages, false);
+ let (_, middle_messages) = streams
+ .poll_messages(&namespace, consumer, args)s file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+use super::bootstrap_test_environment;
use crate::streaming::common::test_setup::TestSetup;
use bytes::BytesMut;
use iggy::prelude::*;
use server::configs::cache_indexes::CacheIndexesConfig;
use server::configs::system::{PartitionConfig, SegmentConfig, SystemConfig};
-use server::streaming::partitions::partition::Partition;
+use server::shard::namespace::IggyFullNamespace;
+use server::shard::system::messages::PollingArgs;
+use server::streaming::polling_consumer::PollingConsumer;
use server::streaming::segments::IggyMessagesBatchMut;
+use server::streaming::traits::MainOps;
use std::collections::HashMap;
use std::str::FromStr;
-use std::sync::Arc;
-use std::sync::atomic::{AtomicU32, AtomicU64};
use test_case::test_matrix;
/*
@@ -91,14 +96,12 @@ async fn test_get_messages_by_offset(
"Running test with message_size: {message_size}, batches: {batch_lengths:?}, messages_required_to_save: {messages_required_to_save}, segment_size: {segment_size}, cache_indexes: {cache_indexes}"
);
+ let shard_id = 0;
let setup = TestSetup::init().await;
- let stream_id = 1;
- let topic_id = 1;
- let partition_id = 1;
let total_messages_count = batch_lengths.iter().sum();
- let config = Arc::new(SystemConfig {
+ let config = SystemConfig {
path: setup.config.path.to_string(),
partition: PartitionConfig {
messages_required_to_save,
@@ -111,28 +114,24 @@ async fn test_get_messages_by_offset(
..Default::default()
},
..Default::default()
- });
+ };
- let mut partition = Partition::create(
- stream_id,
- topic_id,
+ // Use the bootstrap method to create streams with proper slab structure
+ let bootstrap_result = bootstrap_test_environment(shard_id as u16, &config)
+ .await
+ .unwrap();
+ let streams = bootstrap_result.streams;
+ let stream_identifier = bootstrap_result.stream_id;
+ let topic_identifier = bootstrap_result.topic_id;
+ let partition_id = bootstrap_result.partition_id;
+
+ // Create namespace for MainOps calls
+ let namespace = IggyFullNamespace::new(
+ stream_identifier.clone(),
+ topic_identifier.clone(),
partition_id,
- true,
- config.clone(),
- setup.storage.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- IggyTimestamp::now(),
);
- setup.create_partitions_directory(stream_id, topic_id).await;
- partition.persist().await.unwrap();
- partition.open().await.unwrap();
-
let mut all_messages = Vec::with_capacity(total_messages_count as usize);
// Generate all messages as defined in the test matrix
@@ -195,18 +194,33 @@ async fn test_get_messages_by_offset(
let batch = IggyMessagesBatchMut::from_messages(messages_slice_to_append, messages_size);
assert_eq!(batch.count(), batch_len);
- partition.append_messages(batch).await.unwrap();
+ streams
+ .append_messages(shard_id as u16, &config, &namespace, batch)
+ .await
+ .unwrap();
- batch_offsets.push(partition.current_offset);
+ // Get current offset after appending
+ let current_offset = streams.with_partition_by_id(
+ &stream_identifier,
+ &topic_identifier,
+ partition_id,
+ |(_, _, _, offset, ..)| offset.load(std::sync::atomic::Ordering::Relaxed),
+ );
+ batch_offsets.push(current_offset);
current_pos += batch_len as usize;
}
// Use the exact total messages count from the test matrix
let total_sent_messages = total_messages_count;
+ // Create a single consumer to reuse throughout the test
+ let consumer =
+ PollingConsumer::consumer(&Identifier::numeric(1).unwrap(), partition_id as usize);
+
// Test 1: All messages from start
- let all_loaded_messages = partition
- .get_messages_by_offset(0, total_sent_messages)
+ let args = PollingArgs::new(PollingStrategy::offset(0), total_sent_messages, false);
+ let (_, all_loaded_messages) = streams
+ .poll_messages(&namespace, consumer.clone(), args)
.await
.unwrap();
assert_eq!(
@@ -223,8 +237,13 @@ async fn test_get_messages_by_offset(
let prior_batches_sum: u32 = batch_lengths[..3].iter().sum();
let remaining_messages = total_sent_messages - prior_batches_sum;
- let middle_messages = partition
- .get_messages_by_offset(middle_offset + 1, remaining_messages)
+ let args = PollingArgs::new(
+ PollingStrategy::offset(middle_offset + 1),
+ remaining_messages,
+ false,
+ );
+ let (_, middle_messages) = streams
+ .poll_messages(&namespace, consumer.clone(), args)
.await
.unwrap();
@@ -240,8 +259,9 @@ async fn test_get_messages_by_offset(
// Test 3: No messages beyond final offset
if !batch_offsets.is_empty() {
let final_offset = *batch_offsets.last().unwrap();
- let no_messages = partition
- .get_messages_by_offset(final_offset + 1, 1)
+ let args = PollingArgs::new(PollingStrategy::offset(final_offset + 1), 1, false);
+ let (_, no_messages) = streams
+ .poll_messages(&namespace, consumer.clone(), args)
.await
.unwrap();
assert_eq!(
@@ -254,8 +274,9 @@ async fn test_get_messages_by_offset(
// Test 4: Small subset from start
let subset_size = std::cmp::min(3, total_sent_messages);
- let subset_messages = partition
- .get_messages_by_offset(0, subset_size)
+ let args = PollingArgs::new(PollingStrategy::offset(0), subset_size, false);
+ let (_, subset_messages) = streams
+ .poll_messages(&namespace, consumer.clone(), args)
.await
.unwrap();
assert_eq!(
@@ -270,8 +291,9 @@ async fn test_get_messages_by_offset(
if batch_offsets.len() >= 4 {
let span_offset = batch_offsets[1] + 1; // Start from middle of 2nd batch
let span_size = 8; // Should span across 2nd, 3rd, and into 4th batch
- let batches = partition
- .get_messages_by_offset(span_offset, span_size)
+ let args = PollingArgs::new(PollingStrategy::offset(span_offset), span_size, false);
+ let (_, batches) = streams
+ .poll_messages(&namespace, consumer.clone(), args)
.await
.unwrap();
assert_eq!(
@@ -343,8 +365,13 @@ async fn test_get_messages_by_offset(
for chunk_start in (0..total_sent_messages).step_by(chunk_size as usize) {
let read_size = std::cmp::min(chunk_size, total_sent_messages - chunk_start);
- let chunk = partition
- .get_messages_by_offset(chunk_start as u64, read_size)
+ let args = PollingArgs::new(
+ PollingStrategy::offset(chunk_start as u64),
+ read_size,
+ false,
+ );
+ let (_, chunk) = streams
+ .poll_messages(&namespace, consumer.clone(), args)
.await
.unwrap();
diff --git a/core/integration/tests/streaming/get_by_timestamp.rs b/core/integration/tests/streaming/get_by_timestamp.rs
index f0a952ec..a40d2904 100644
--- a/core/integration/tests/streaming/get_by_timestamp.rs
+++ b/core/integration/tests/streaming/get_by_timestamp.rs
@@ -17,12 +17,16 @@
*/
use crate::streaming::common::test_setup::TestSetup;
+use super::bootstrap_test_environment;
use bytes::BytesMut;
use iggy::prelude::*;
use server::configs::cache_indexes::CacheIndexesConfig;
use server::configs::system::{PartitionConfig, SegmentConfig, SystemConfig};
-use server::streaming::partitions::partition::Partition;
+use server::shard::system::messages::PollingArgs;
+use server::streaming::polling_consumer::PollingConsumer;
use server::streaming::segments::IggyMessagesBatchMut;
+use server::streaming::traits::MainOps;
+use server::shard::namespace::IggyFullNamespace;
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Arc;
@@ -92,14 +96,12 @@ async fn test_get_messages_by_timestamp(
"Running test with message_size: {message_size}, batches: {batch_lengths:?}, messages_required_to_save: {messages_required_to_save}, segment_size: {segment_size}, cache_indexes: {cache_indexes}"
);
+ let shard_id = 0;
let setup = TestSetup::init().await;
- let stream_id = 1;
- let topic_id = 1;
- let partition_id = 1;
let total_messages_count = batch_lengths.iter().sum();
- let config = Arc::new(SystemConfig {
+ let config = SystemConfig {
path: setup.config.path.to_string(),
partition: PartitionConfig {
messages_required_to_save,
@@ -112,27 +114,17 @@ async fn test_get_messages_by_timestamp(
..Default::default()
},
..Default::default()
- });
-
- let mut partition = Partition::create(
- stream_id,
- topic_id,
- partition_id,
- true,
- config.clone(),
- setup.storage.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- IggyTimestamp::now(),
- );
-
- setup.create_partitions_directory(stream_id, topic_id).await;
- partition.persist().await.unwrap();
- partition.open().await.unwrap();
+ };
+
+ // Use the bootstrap method to create streams with proper slab structure
+ let bootstrap_result = bootstrap_test_environment(shard_id as u16, &config).await.unwrap();
+ let streams = bootstrap_result.streams;
+ let stream_identifier = bootstrap_result.stream_id;
+ let topic_identifier = bootstrap_result.topic_id;
+ let partition_id = bootstrap_result.partition_id;
+
+ // Create namespace for MainOps calls
+ let namespace = IggyFullNamespace::new(stream_identifier.clone(), topic_identifier.clone(), partition_id);
let mut all_messages = Vec::with_capacity(total_messages_count as usize);
@@ -201,7 +193,7 @@ async fn test_get_messages_by_timestamp(
let batch = IggyMessagesBatchMut::from_messages(messages_slice_to_append, messages_size);
assert_eq!(batch.count(), batch_len);
- partition.append_messages(batch).await.unwrap();
+ streams.append_messages(shard_id as u16, &config, &namespace, batch).await.unwrap();
// Capture the timestamp of this batch
batch_timestamps.push(IggyTimestamp::now());
@@ -216,9 +208,13 @@ async fn test_get_messages_by_timestamp(
// Use the exact total messages count from the test matrix
let total_sent_messages = total_messages_count;
+ // Create a single consumer to reuse throughout the test
+ let consumer = PollingConsumer::consumer(&Identifier::numeric(1).unwrap(), partition_id as usize);
+
// Test 1: All messages from initial timestamp
- let all_loaded_messages = partition
- .get_messages_by_timestamp(initial_timestamp, total_sent_messages)
+ let args = PollingArgs::new(PollingStrategy::timestamp(initial_timestamp), total_sent_messages, false);
+ let (_, all_loaded_messages) = streams
+ .poll_messages(&namespace, consumer, args)
.await
.unwrap();
assert_eq!(
@@ -239,8 +235,9 @@ async fn test_get_messages_by_timestamp(
let prior_batches_sum: u32 = batch_lengths[..3].iter().sum();
let remaining_messages = total_sent_messages - prior_batches_sum;
- let middle_messages = partition
- .get_messages_by_timestamp(middle_timestamp, remaining_messages)
+ let args = PollingArgs::new(PollingStrategy::timestamp(middle_timestamp), remaining_messages, false);
+ let (_, middle_messages) = streams
+ .poll_messages(&namespace, consumer, args)
.await
.unwrap();
@@ -254,8 +251,9 @@ async fn test_get_messages_by_timestamp(
}
// Test 3: No messages after final timestamp
- let no_messages = partition
- .get_messages_by_timestamp(final_timestamp, 1)
+ let args = PollingArgs::new(PollingStrategy::timestamp(final_timestamp), 1, false);
+ let (_, no_messages) = streams
+ .poll_messages(&namespace, consumer, args)
.await
.unwrap();
assert_eq!(
@@ -267,8 +265,9 @@ async fn test_get_messages_by_timestamp(
// Test 4: Small subset from initial timestamp
let subset_size = std::cmp::min(3, total_sent_messages);
- let subset_messages = partition
- .get_messages_by_timestamp(initial_timestamp, subset_size)
+ let args = PollingArgs::new(PollingStrategy::timestamp(initial_timestamp), subset_size, false);
+ let (_, subset_messages) = streams
+ .poll_messages(&namespace, consumer, args)
.await
.unwrap();
assert_eq!(
@@ -285,8 +284,9 @@ async fn test_get_messages_by_timestamp(
let span_timestamp = IggyTimestamp::from(batch_timestamps[1].as_micros() + 1000);
let span_size = 8; // Should span across multiple batches
- let spanning_messages = partition
- .get_messages_by_timestamp(span_timestamp, span_size)
+ let args = PollingArgs::new(PollingStrategy::timestamp(span_timestamp), span_size, false);
+ let (_, spanning_messages) = streams
+ .poll_messages(&namespace, consumer, args)
.await
.unwrap();
diff --git a/core/integration/tests/streaming/messages.rs b/core/integration/tests/streaming/messages.rs
deleted file mode 100644
index 3561dd7b..00000000
--- a/core/integration/tests/streaming/messages.rs
+++ /dev/null
@@ -1,164 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::common::test_setup::TestSetup;
-use bytes::Bytes;
-use iggy::prelude::*;
-use server::configs::system::{PartitionConfig, SystemConfig};
-use server::state::system::PartitionState;
-use server::streaming::partitions::partition::Partition;
-use server::streaming::segments::IggyMessagesBatchMut;
-use std::collections::HashMap;
-use std::str::FromStr;
-use std::sync::Arc;
-use std::sync::atomic::{AtomicU32, AtomicU64};
-
-#[compio::test]
-async fn should_persist_messages_and_then_load_them_from_disk() {
- let setup = TestSetup::init().await;
- let stream_id = 1;
- let topic_id = 1;
- let partition_id = 1;
- let messages_count = 1000;
- let config = Arc::new(SystemConfig {
- path: setup.config.path.to_string(),
- partition: PartitionConfig {
- messages_required_to_save: messages_count,
- enforce_fsync: true,
- ..Default::default()
- },
- ..Default::default()
- });
- let mut partition = Partition::create(
- stream_id,
- topic_id,
- partition_id,
- true,
- config.clone(),
- setup.storage.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- IggyTimestamp::now(),
- );
- partition.persist().await.unwrap();
- partition.open().await.unwrap();
-
- let mut messages = Vec::with_capacity(messages_count as usize);
- let mut appended_messages = Vec::with_capacity(messages_count as usize);
- for i in 1..=messages_count {
- let id = i as u128;
- let payload = Bytes::from(format!("message {i}"));
- let mut headers = HashMap::new();
- headers.insert(
- HeaderKey::new("key_1").unwrap(),
- HeaderValue::from_str("Value 1").unwrap(),
- );
- headers.insert(
- HeaderKey::new("key 2").unwrap(),
- HeaderValue::from_bool(true).unwrap(),
- );
- headers.insert(
- HeaderKey::new("key-3").unwrap(),
- HeaderValue::from_uint64(123456).unwrap(),
- );
-
- let appended_message = IggyMessage::builder()
- .id(id)
- .payload(payload.clone())
- .user_headers(headers.clone())
- .build()
- .expect("Failed to create message with headers");
- let message = IggyMessage::builder()
- .id(id)
- .payload(payload)
- .user_headers(headers)
- .build()
- .expect("Failed to create message with headers");
-
- appended_messages.push(appended_message);
- messages.push(message);
- }
-
- setup.create_partitions_directory(stream_id, topic_id).await;
- partition.persist().await.unwrap();
- let messages_size = messages
- .iter()
- .map(|msg| msg.get_size_bytes().as_bytes_u32())
- .sum::<u32>();
- let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition.append_messages(batch).await.unwrap();
- assert_eq!(
- partition.unsaved_messages_count, 0,
- "Expected unsaved messages count to be 0, but got {}",
- partition.unsaved_messages_count
- );
-
- let now = IggyTimestamp::now();
- let mut loaded_partition = Partition::create(
- stream_id,
- topic_id,
- partition.partition_id,
- false,
- config.clone(),
- setup.storage.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- now,
- );
- let partition_state = PartitionState {
- id: partition.partition_id,
- created_at: now,
- };
- loaded_partition.load(partition_state).await.unwrap();
- let loaded_messages = loaded_partition
- .get_messages_by_offset(0, messages_count)
- .await
- .unwrap();
- assert_eq!(
- loaded_messages.count(),
- messages_count,
- "Unexpected loaded messages count "
- );
- (0..messages_count as usize).for_each(|i| {
- let loaded_message = &loaded_messages.get(i).unwrap();
- let appended_message = &appended_messages[i];
- assert_eq!(
- loaded_message.header().id(),
- appended_message.header.id,
- "Message ID mismatch at position {i}",
- );
- assert_eq!(
- loaded_message.payload(),
- appended_message.payload,
- "Payload mismatch at position {i}",
- );
- assert_eq!(
- loaded_message.user_headers_map().unwrap().unwrap(),
- appended_message.user_headers_map().unwrap().unwrap(),
- "Headers mismatch at position {i}",
- );
- });
-}
diff --git a/core/integration/tests/streaming/mod.rs b/core/integration/tests/streaming/mod.rs
index f51b875b..4d547be7 100644
--- a/core/integration/tests/streaming/mod.rs
+++ b/core/integration/tests/streaming/mod.rs
@@ -18,19 +18,113 @@
use bytes::Bytes;
use iggy::prelude::IggyMessage;
+use iggy_common::{CompressionAlgorithm, Identifier, IggyError, IggyExpiry, MaxTopicSize};
+use server::{
+ configs::system::SystemConfig,
+ slab::{streams::Streams, traits_ext::EntityMarker},
+ streaming::{
+ self,
+ partitions::{partition2, storage2::create_partition_file_hierarchy},
+ segments::{Segment2, storage::create_segment_storage},
+ streams::{storage2::create_stream_file_hierarchy, stream2},
+ topics::{storage2::create_topic_file_hierarchy, topic2},
+ },
+};
mod common;
-mod consumer_offset;
mod get_by_offset;
mod get_by_timestamp;
-mod messages;
-mod partition;
-mod segment;
-mod shard;
mod snapshot;
-mod stream;
-mod topic;
-mod topic_messages;
+
+struct BootstrapResult {
+ streams: Streams,
+ stream_id: Identifier,
+ topic_id: Identifier,
+ partition_id: usize,
+}
+
+async fn bootstrap_test_environment(
+ shard_id: u16,
+ config: &SystemConfig,
+) -> Result<BootstrapResult, IggyError> {
+ let stream_name = "stream-1".to_owned();
+ let topic_name = "topic-1".to_owned();
+ let topic_expiry = IggyExpiry::NeverExpire;
+ let topic_size = MaxTopicSize::Unlimited;
+ let partitions_count = 1;
+
+ let streams = Streams::default();
+ // Create stream together with its dirs
+ let stream = stream2::create_and_insert_stream_mem(&streams, stream_name);
+ create_stream_file_hierarchy(shard_id, stream.id(), &config).await?;
+ // Create topic together with its dirs
+ let stream_id = Identifier::numeric(stream.id() as u32).unwrap();
+ let parent_stats = streams.with_stream_by_id(&stream_id, |(_, stats)| stats.clone());
+ let message_expiry = config.resolve_message_expiry(topic_expiry);
+ let max_topic_size = config.resolve_max_topic_size(topic_size)?;
+
+ let topic = topic2::create_and_insert_topics_mem(
+ &streams,
+ &stream_id,
+ topic_name,
+ 1,
+ message_expiry,
+ CompressionAlgorithm::default(),
+ max_topic_size,
+ parent_stats,
+ );
+ create_topic_file_hierarchy(shard_id, stream.id(), topic.id(), &config).await?;
+ // Create partition together with its dirs
+ let topic_id = Identifier::numeric(topic.id() as u32).unwrap();
+ let parent_stats = streams.with_topic_by_id(
+ &stream_id,
+ &topic_id,
+ streaming::topics::helpers::get_stats(),
+ );
+ let partitions = partition2::create_and_insert_partitions_mem(
+ &streams,
+ &stream_id,
+ &topic_id,
+ parent_stats,
+ partitions_count,
+ config,
+ );
+ for partition in partitions {
+ create_partition_file_hierarchy(shard_id, stream.id(), topic.id(), partition.id(), &config)
+ .await?;
+
+ // Open the log
+ let start_offset = 0;
+ let segment = Segment2::new(
+ start_offset,
+ config.segment.size,
+ config.segment.message_expiry,
+ );
+ let messages_size = 0;
+ let indexes_size = 0;
+ let storage = create_segment_storage(
+ &config,
+ stream.id(),
+ topic.id(),
+ partition.id(),
+ messages_size,
+ indexes_size,
+ start_offset,
+ )
+ .await?;
+
+ streams.with_partition_by_id_mut(&stream_id, &topic_id, partition.id(), |(.., log)| {
+ log.add_persisted_segment(segment, storage);
+ });
+ }
+
+ Ok(BootstrapResult {
+ streams,
+ stream_id,
+ topic_id,
+ partition_id: 0,
+ })
+}
fn create_messages() -> Vec<IggyMessage> {
vec![
diff --git a/core/integration/tests/streaming/partition.rs b/core/integration/tests/streaming/partition.rs
deleted file mode 100644
index ac657e3b..00000000
--- a/core/integration/tests/streaming/partition.rs
+++ /dev/null
@@ -1,225 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::common::test_setup::TestSetup;
-use crate::streaming::create_messages;
-use compio::fs;
-use iggy::prelude::{IggyExpiry, IggyTimestamp, Sizeable};
-use server::state::system::PartitionState;
-use server::streaming::partitions::partition::Partition;
-use server::streaming::segments::*;
-use std::sync::Arc;
-use std::sync::atomic::{AtomicU32, AtomicU64};
-
-#[compio::test]
-async fn should_persist_partition_with_segment() {
- let setup = TestSetup::init().await;
- let with_segment = true;
- let stream_id = 1;
- let topic_id = 2;
- setup.create_partitions_directory(stream_id, topic_id).await;
- let partition_ids = get_partition_ids();
- for partition_id in partition_ids {
- let mut partition = Partition::create(
- stream_id,
- topic_id,
- partition_id,
- with_segment,
- setup.config.clone(),
- setup.storage.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- IggyTimestamp::now(),
- );
-
- partition.persist().await.unwrap();
- partition.open().await.unwrap();
-
- assert_persisted_partition(&partition.partition_path, with_segment).await;
- }
-}
-
-#[compio::test]
-async fn should_load_existing_partition_from_disk() {
- let setup = TestSetup::init().await;
- let with_segment = true;
- let stream_id = 1;
- let topic_id = 2;
- setup.create_partitions_directory(stream_id, topic_id).await;
- let partition_ids = get_partition_ids();
- for partition_id in partition_ids {
- let mut partition = Partition::create(
- stream_id,
- topic_id,
- partition_id,
- with_segment,
- setup.config.clone(),
- setup.storage.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- IggyTimestamp::now(),
- );
- partition.persist().await.unwrap();
- partition.open().await.unwrap();
- assert_persisted_partition(&partition.partition_path, with_segment).await;
-
- let now = IggyTimestamp::now();
- let mut loaded_partition = Partition::create(
- stream_id,
- topic_id,
- partition.partition_id,
- false,
- setup.config.clone(),
- setup.storage.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- now,
- );
- let partition_state = PartitionState {
- id: partition.partition_id,
- created_at: now,
- };
- loaded_partition.load(partition_state).await.unwrap();
-
- assert_eq!(loaded_partition.stream_id, partition.stream_id);
- assert_eq!(loaded_partition.partition_id, partition.partition_id);
- assert_eq!(loaded_partition.partition_path, partition.partition_path);
- assert_eq!(loaded_partition.current_offset, partition.current_offset);
- assert_eq!(
- loaded_partition.unsaved_messages_count,
- partition.unsaved_messages_count
- );
- assert_eq!(
- loaded_partition.get_segments().len(),
- partition.get_segments().len()
- );
- assert_eq!(
- loaded_partition.should_increment_offset,
- partition.should_increment_offset
- );
- }
-}
-
-#[compio::test]
-async fn should_delete_existing_partition_from_disk() {
- let setup = TestSetup::init().await;
- let with_segment = true;
- let stream_id = 1;
- let topic_id = 2;
- setup.create_partitions_directory(stream_id, topic_id).await;
- let partition_ids = get_partition_ids();
- for partition_id in partition_ids {
- let mut partition = Partition::create(
- stream_id,
- topic_id,
- partition_id,
- with_segment,
- setup.config.clone(),
- setup.storage.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- IggyTimestamp::now(),
- );
- partition.persist().await.unwrap();
- partition.open().await.unwrap();
- assert_persisted_partition(&partition.partition_path, with_segment).await;
-
- partition.delete().await.unwrap();
-
- assert!(fs::metadata(&partition.partition_path).await.is_err());
- }
-}
-
-#[compio::test]
-async fn should_purge_existing_partition_on_disk() {
- let setup = TestSetup::init().await;
- let with_segment = true;
- let stream_id = 1;
- let topic_id = 2;
- setup.create_partitions_directory(stream_id, topic_id).await;
- let partition_ids = get_partition_ids();
- for partition_id in partition_ids {
- let mut partition = Partition::create(
- stream_id,
- topic_id,
- partition_id,
- with_segment,
- setup.config.clone(),
- setup.storage.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- IggyTimestamp::now(),
- );
- partition.persist().await.unwrap();
- partition.open().await.unwrap();
- assert_persisted_partition(&partition.partition_path, with_segment).await;
- let messages = create_messages();
- let messages_count = messages.len() as u32;
- let messages_size: u32 = messages
- .iter()
- .map(|msg| msg.get_size_bytes().as_bytes_u32())
- .sum();
- let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition.append_messages(batch).await.unwrap();
- let loaded_messages = partition.get_messages_by_offset(0, 100).await.unwrap();
- assert_eq!(loaded_messages.count(), messages_count);
- partition.purge().await.unwrap();
- assert_eq!(partition.current_offset, 0);
- assert_eq!(partition.unsaved_messages_count, 0);
- assert!(!partition.should_increment_offset);
- let loaded_messages = partition.get_messages_by_offset(0, 100).await.unwrap();
- assert!(loaded_messages.is_empty());
- }
-}
-
-async fn assert_persisted_partition(partition_path: &str, with_segment: bool) {
- assert!(fs::metadata(&partition_path).await.is_ok());
-
- if with_segment {
- let start_offset = 0u64;
- let segment_path = format!("{partition_path}/{start_offset:0>20}");
- let messages_file_path = format!("{segment_path}.{LOG_EXTENSION}");
- let index_path = format!("{segment_path}.{INDEX_EXTENSION}");
- assert!(fs::metadata(&messages_file_path).await.is_ok());
- assert!(fs::metadata(&index_path).await.is_ok());
- }
-}
-
-fn get_partition_ids() -> Vec<u32> {
- vec![1, 2, 3, 5, 10, 100, 1000, 99999]
-}
diff --git a/core/integration/tests/streaming/segment.rs b/core/integration/tests/streaming/segment.rs
deleted file mode 100644
index 9a92e7fe..00000000
--- a/core/integration/tests/streaming/segment.rs
+++ /dev/null
@@ -1,575 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::common::test_setup::TestSetup;
-use crate::streaming::create_message;
-use bytes::Bytes;
-use compio::fs;
-use compio::time::sleep;
-use iggy::prelude::locking::IggyRwLockFn;
-use iggy::prelude::*;
-use server::configs::cluster::ClusterConfig;
-use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig};
-use server::configs::system::{PartitionConfig, SegmentConfig, SystemConfig};
-use server::shard::IggyShard;
-use server::streaming::segments::*;
-use server::streaming::session::Session;
-use std::fs::DirEntry;
-use std::net::{Ipv4Addr, SocketAddr};
-use std::rc::Rc;
-use std::str::FromStr;
-use std::sync::Arc;
-use std::sync::atomic::AtomicU64;
-use std::time::Duration;
-
-#[compio::test]
-async fn should_persist_segment() {
- let setup = TestSetup::init().await;
- let stream_id = 1;
- let topic_id = 2;
- let partition_id = 3;
- let start_offsets = get_start_offsets();
- for start_offset in start_offsets {
- let mut segment = Segment::create(
- stream_id,
- topic_id,
- partition_id,
- start_offset,
- setup.config.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- true,
- );
-
- setup
- .create_partition_directory(stream_id, topic_id, partition_id)
- .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,
- ),
- start_offset,
- )
- .await;
- }
-}
-
-#[compio::test]
-async fn should_load_existing_segment_from_disk() {
- let setup = TestSetup::init().await;
- let stream_id = 1;
- let topic_id = 2;
- let partition_id = 3;
- let start_offsets = get_start_offsets();
- for start_offset in start_offsets {
- let mut segment = Segment::create(
- stream_id,
- topic_id,
- partition_id,
- start_offset,
- setup.config.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- true,
- );
- setup
- .create_partition_directory(stream_id, topic_id, partition_id)
- .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,
- ),
- start_offset,
- )
- .await;
-
- let mut loaded_segment = Segment::create(
- stream_id,
- topic_id,
- partition_id,
- start_offset,
- setup.config.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- false,
- );
- loaded_segment.load_from_disk().await.unwrap();
- let loaded_messages = loaded_segment.get_messages_by_offset(0, 10).await.unwrap();
-
- assert_eq!(loaded_segment.partition_id(), segment.partition_id());
- assert_eq!(loaded_segment.start_offset(), segment.start_offset());
- assert_eq!(loaded_segment.end_offset(), segment.end_offset());
- assert_eq!(
- loaded_segment.get_messages_size(),
- segment.get_messages_size()
- );
- assert_eq!(loaded_segment.is_closed(), segment.is_closed());
- assert_eq!(
- loaded_segment.messages_file_path(),
- segment.messages_file_path()
- );
- assert_eq!(loaded_segment.index_file_path(), segment.index_file_path());
- assert!(loaded_messages.is_empty());
- }
-}
-
-#[compio::test]
-async fn should_persist_and_load_segment_with_messages() {
- let setup = TestSetup::init().await;
- let stream_id = 1;
- let topic_id = 2;
- let partition_id = 3;
- let start_offset = 0;
- let mut segment = Segment::create(
- stream_id,
- topic_id,
- partition_id,
- start_offset,
- setup.config.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- true,
- );
-
- setup
- .create_partition_directory(stream_id, topic_id, partition_id)
- .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,
- ),
- start_offset,
- )
- .await;
- let messages_count = 10;
- let mut messages = Vec::new();
- let mut messages_size = 0;
- for i in 0..messages_count {
- let message = IggyMessage::builder()
- .id(i as u128)
- .payload(Bytes::from("test"))
- .build()
- .expect("Failed to create message");
- messages_size += message.get_size_bytes().as_bytes_u32();
- messages.push(message);
- }
- let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
-
- segment.append_batch(0, batch, None).await.unwrap();
- segment.persist_messages().await.unwrap();
- let mut loaded_segment = Segment::create(
- stream_id,
- topic_id,
- partition_id,
- start_offset,
- setup.config.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- false,
- );
- loaded_segment.load_from_disk().await.unwrap();
- let messages = loaded_segment
- .get_messages_by_offset(0, messages_count)
- .await
- .unwrap();
- assert_eq!(messages.count(), messages_count);
-}
-
-#[compio::test]
-async fn given_at_least_one_not_expired_message_segment_should_not_be_expired() {
- let config = SystemConfig {
- partition: PartitionConfig {
- enforce_fsync: true,
- ..Default::default()
- },
- segment: SegmentConfig {
- size: IggyByteSize::from_str("10B").unwrap(), // small size to force expiration
- ..Default::default()
- },
- ..Default::default()
- };
- let setup = TestSetup::init_with_config(config).await;
- let stream_id = 1;
- let topic_id = 2;
- let partition_id = 3;
- let start_offset = 0;
- let message_expiry_us = 50000;
- let message_expiry = message_expiry_us.into();
- let mut segment = Segment::create(
- stream_id,
- topic_id,
- partition_id,
- start_offset,
- setup.config.clone(),
- message_expiry,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- true,
- );
-
- setup
- .create_partition_directory(stream_id, topic_id, partition_id)
- .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,
- ),
- start_offset,
- )
- .await;
-
- let nothing_expired_ts = IggyTimestamp::now();
-
- let first_message = vec![
- IggyMessage::builder()
- .payload(Bytes::from("expired"))
- .build()
- .expect("Failed to create message"),
- ];
- let first_message_size = first_message[0].get_size_bytes().as_bytes_u32();
- let first_batch = IggyMessagesBatchMut::from_messages(&first_message, first_message_size);
- segment.append_batch(0, first_batch, None).await.unwrap();
-
- sleep(Duration::from_micros(message_expiry_us / 2)).await;
- let first_message_expired_ts = IggyTimestamp::now();
-
- let second_message = vec![
- IggyMessage::builder()
- .payload(Bytes::from("not-expired"))
- .build()
- .expect("Failed to create message"),
- ];
- let second_message_size = second_message[0].get_size_bytes().as_bytes_u32();
- let second_batch = IggyMessagesBatchMut::from_messages(&second_message, second_message_size);
- segment.append_batch(1, second_batch, None).await.unwrap();
-
- let second_message_expired_ts =
- IggyTimestamp::now() + IggyDuration::from(message_expiry_us * 2);
-
- segment.persist_messages().await.unwrap();
-
- assert!(
- !segment.is_expired(nothing_expired_ts).await,
- "Segment should not be expired for nothing expired timestamp"
- );
- assert!(
- !segment.is_expired(first_message_expired_ts).await,
- "Segment should not be expired for first message expired timestamp"
- );
- assert!(
- segment.is_expired(second_message_expired_ts).await,
- "Segment should be expired for second message expired timestamp"
- );
-}
-
-#[compio::test]
-async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Error>> {
- let config = SystemConfig {
- segment: SegmentConfig {
- size: IggyByteSize::from_str("10B").unwrap(), // small size to force segment closure
- ..Default::default()
- },
- ..Default::default()
- };
- let setup = TestSetup::init_with_config(config).await;
- let mut system = System::new(
- setup.config.clone(),
- ClusterConfig::default(),
- DataMaintenanceConfig::default(),
- PersonalAccessTokenConfig::default(),
- );
-
- // Properties
- let stream_id = Identifier::numeric(1)?;
- let stream_name = "test";
- let topic_name = "test_topic";
- let topic_id = Identifier::numeric(1)?;
- let partition_id = 1;
-
- setup
- .create_partition_directory(
- stream_id.get_u32_value()?,
- topic_id.get_u32_value()?,
- partition_id,
- )
- .await;
-
- let session = Session::new(1, 1, SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234));
- let session = Rc::new(session);
- shard.add_active_session(session.clone());
-
- let id = shard
- .create_stream_bypass_auth(Some(stream_id.get_u32_value()?), stream_name)
- .unwrap();
- let stream = shard.get_stream(&Identifier::numeric(id).unwrap()).unwrap();
- stream.persist();
- drop(stream);
-
- let result = shard
- .create_topic(
- &session,
- &stream_id,
- Some(topic_id.get_u32_value()?),
- topic_name,
- partition_id,
- IggyExpiry::default(),
- CompressionAlgorithm::default(),
- MaxTopicSize::default(),
- None,
- )
- .await;
- if let Some(error) = result.err() {
- error!("Failed to create topic: {}", error);
- }
-
- let stream = shard.find_stream(&session, &stream_id)?;
- let topic = shard.find_topic(&session, &stream, &topic_id)?;
- let partitions = topic.get_partitions();
- for partition in &partitions {
- let mut partition_lock = partition.write().await;
- partition_lock.persist().await.unwrap();
- partition_lock.open().await.unwrap();
- drop(partition_lock);
- }
- let partition = partitions
- .first()
- .ok_or(IggyError::Error)
- .inspect_err(|_| log::error!("Cannot retrieve initial partition."))?;
-
- // Create multiple segments by sending messages one by one
- // Each message should be large enough to cause a segment to close
- for i in 0..5 {
- let mut partition_lock = partition.write().await;
-
- let payload = "payload that will force segment closure due to size";
- let message = create_message(i, payload);
- let messages = vec![message];
-
- let messages_size = messages
- .iter()
- .map(|m| m.get_size_bytes().as_bytes_u32())
- .sum();
- let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
-
- partition_lock.append_messages(batch).await.unwrap();
-
- drop(partition_lock);
-
- sleep(Duration::from_millis(100)).await;
- }
- drop(stream);
-
- let partition_path = setup.config.get_partition_path(
- 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);
- println!(
- "Created segments: {:?}",
- initial_segments
- .iter()
- .map(|e| e.file_name())
- .collect::<Vec<_>>()
- );
-
- // Since we've created 5 messages with large payloads, we should have multiple segments
- assert!(
- initial_segments.len() >= 4,
- "Expected at least 4 segments but got {}",
- initial_segments.len()
- );
-
- let mut initial_offsets = get_segment_offsets(&initial_segments);
- initial_offsets.sort();
- println!("Initial segment offsets (sorted): {initial_offsets:?}");
-
- let first_keep_count = 3usize;
- shard
- .delete_segments(
- &session,
- &stream_id,
- &topic_id,
- partition_id,
- first_keep_count as u32,
- )
- .await?;
-
- let segments_after_first_delete = get_segment_paths_for_partition(&partition_path);
- println!(
- "Segments after first deletion (keep={}): {:?}",
- first_keep_count,
- segments_after_first_delete
- .iter()
- .map(|e| e.file_name())
- .collect::<Vec<_>>()
- );
-
- // We should have at most first_keep_count + 1 segments after first deletion
- // (keep_count closed segments + 1 open segment)
- assert!(
- segments_after_first_delete.len() <= first_keep_count + 1,
- "Should have at most {} segments after first deletion but got {}",
- first_keep_count + 1,
- segments_after_first_delete.len()
- );
-
- // Second attempt to delete segments - keeping only 1 closed segment + 1 open
- let second_keep_count = 1usize;
- shard
- .delete_segments(
- &session,
- &stream_id,
- &topic_id,
- partition_id,
- second_keep_count as u32,
- )
- .await?;
-
- let segments_after_second_delete = get_segment_paths_for_partition(&partition_path);
- println!(
- "Segments after second deletion (keep={}): {:?}",
- second_keep_count,
- segments_after_second_delete
- .iter()
- .map(|e| e.file_name())
- .collect::<Vec<_>>()
- );
-
- // We should have at most second_keep_count + 1 segments after second deletion
- // (keep_count closed segments + 1 open segment)
- assert!(
- segments_after_second_delete.len() <= second_keep_count + 1,
- "Should have at most {} segments after second deletion but got {}",
- second_keep_count + 1,
- segments_after_second_delete.len()
- );
-
- let mut final_offsets = get_segment_offsets(&segments_after_second_delete);
- final_offsets.sort();
-
- let highest_initial_offsets: Vec<u64> = initial_offsets
- .iter()
- .rev()
- .take(second_keep_count + 1)
- .copied()
- .collect();
-
- for &offset in &final_offsets {
- assert!(
- highest_initial_offsets.contains(&offset),
- "Offset {offset} should not remain after final deletion"
- );
- }
-
- Ok(())
-}
-
-// Helper function to extract segment offsets from DirEntry list
-fn get_segment_offsets(segments: &[DirEntry]) -> Vec<u64> {
- segments
- .iter()
- .filter_map(|entry| {
- entry.file_name().to_str().and_then(|name| {
- name.split('.')
- .next()
- .and_then(|offset_str| offset_str.parse::<u64>().ok())
- })
- })
- .collect()
-}
-
-async fn assert_persisted_segment(partition_path: &str, start_offset: u64) {
- let segment_path = format!("{partition_path}/{start_offset:0>20}");
- let messages_file_path = format!("{segment_path}.{LOG_EXTENSION}");
- let index_path = format!("{segment_path}.{INDEX_EXTENSION}");
- assert!(fs::metadata(&messages_file_path).await.is_ok());
- assert!(fs::metadata(&index_path).await.is_ok());
-}
-
-fn get_segment_paths_for_partition(partition_path: &str) -> Vec<DirEntry> {
- std::fs::read_dir(partition_path)
- .map(|read_dir| {
- read_dir
- .filter_map(|dir_entry| {
- dir_entry
- .map(|dir_entry| {
- match dir_entry
- .path()
- .extension()
- .is_some_and(|ext| ext == LOG_EXTENSION)
- {
- true => Some(dir_entry),
- false => None,
- }
- })
- .ok()
- .flatten()
- })
- .collect::<Vec<_>>()
- })
- .unwrap_or_default()
-}
-
-fn get_start_offsets() -> Vec<u64> {
- vec![
- 0, 1, 2, 9, 10, 99, 100, 110, 200, 1000, 1234, 12345, 100000, 9999999,
- ]
-}
diff --git a/core/integration/tests/streaming/shard.rs b/core/integration/tests/streaming/shard.rs
deleted file mode 100644
index ec2e26da..00000000
--- a/core/integration/tests/streaming/shard.rs
+++ /dev/null
@@ -1,170 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::common::test_setup::TestSetup;
-use compio::fs;
-use iggy::prelude::Identifier;
-use iggy_common::locking::IggyRwLockFn;
-use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig, ServerConfig};
-use server::configs::system::SystemConfig;
-use server::shard::IggyShard;
-use server::streaming::session::Session;
-use std::net::{Ipv4Addr, SocketAddr};
-use std::rc::Rc;
-
-//TODO: Fix me use shard instead of system
-#[compio::test]
-async fn should_initialize_system_and_base_directories() {
- let setup = TestSetup::init().await;
- let server_config = ServerConfig {
- system: setup.config.clone(),
- ..Default::default()
- };
- let shard = IggyShard::default_from_config(server_config);
- shard.init().await.unwrap();
-
- let mut dir_entries = server::io::fs_utils::walk_dir(&setup.config.path)
- .await
- .unwrap();
- let mut names = Vec::new();
- for dir_entry in dir_entries.iter_mut() {
- if dir_entry.is_dir {
- names.push(dir_entry.name.clone().unwrap());
- } else {
- dir_entry.name = Some(
- dir_entry
- .path
- .file_name()
- .unwrap()
- .to_string_lossy()
- .to_string(),
- );
- }
- }
-
- // Root + streams + state + runtime
- assert_eq!(names.len(), 4);
- let mut has_stream = false;
- for name in names {
- if name.contains(&setup.config.stream.path) {
- has_stream = true;
- }
- }
- assert!(has_stream);
-}
-
-#[compio::test]
-async fn should_create_and_persist_stream() {
- let setup = TestSetup::init().await;
- let server_config = ServerConfig {
- system: setup.config.clone(),
- ..Default::default()
- };
- let shard = IggyShard::default_from_config(server_config);
- let stream_id = 1;
- let stream_name = "test";
- let session = Session::new(1, 1, SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234));
- let session = Rc::new(session);
- shard.init().await.unwrap();
- shard.add_active_session(session.clone());
-
- shard
- .create_stream_bypass_auth(Some(stream_id), stream_name)
- .unwrap();
- let stream = shard
- .get_stream(&Identifier::numeric(stream_id).unwrap())
- .unwrap();
- stream.persist().await.unwrap();
-
- assert_persisted_stream(&setup.config.get_streams_path(), stream_id).await;
-}
-
-#[compio::test]
-async fn should_create_and_persist_stream_with_automatically_generated_id() {
- let setup = TestSetup::init().await;
- let server_config = ServerConfig {
- system: setup.config.clone(),
- ..Default::default()
- };
- let shard = IggyShard::default_from_config(server_config);
- let stream_id = 1;
- let stream_name = "test";
- shard.init().await.unwrap();
- let session = Session::new(1, 1, SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234));
- let session = Rc::new(session);
- shard.add_active_session(session.clone());
-
- let stream_id = shard.create_stream_bypass_auth(None, stream_name).unwrap();
- let stream = shard
- .get_stream(&Identifier::numeric(stream_id).unwrap())
- .unwrap();
- stream.persist().await.unwrap();
-
- assert_persisted_stream(&setup.config.get_streams_path(), stream_id).await;
-}
-
-#[compio::test]
-async fn should_delete_persisted_stream() {
- let setup = TestSetup::init().await;
- let server_config = ServerConfig {
- system: setup.config.clone(),
- ..Default::default()
- };
- let shard = IggyShard::default_from_config(server_config);
- let stream_id = 1;
- let stream_name = "test";
- let session = Session::new(1, 1, SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234));
- let session = Rc::new(session);
- shard.init().await.unwrap();
- shard.add_active_session(session.clone());
-
- let id = shard
- .create_stream_bypass_auth(Some(stream_id), stream_name)
- .unwrap();
- let stream = shard.get_stream(&Identifier::numeric(id).unwrap()).unwrap();
- stream.persist().await.unwrap();
- drop(stream);
- assert_persisted_stream(&setup.config.get_streams_path(), stream_id).await;
- let stream_path = shard
- .get_stream(&Identifier::numeric(stream_id).unwrap())
- .unwrap()
- .path
- .clone();
-
- let stream = shard
- .delete_stream_bypass_auth(&Identifier::numeric(stream_id).unwrap())
- .unwrap();
- for topic in stream.get_topics() {
- let partitions = topic.get_partitions();
- for partition in partitions {
- let mut part = partition.write().await;
- part.delete().await.unwrap();
- }
- topic.delete().await.unwrap();
- }
- stream.delete().await.unwrap();
- assert!(fs::metadata(stream_path).await.is_err());
-}
-
-async fn assert_persisted_stream(streams_path: &str, stream_id: u32) {
- let streams_metadata = fs::metadata(streams_path).await.unwrap();
- assert!(streams_metadata.is_dir());
- let stream_path = format!("{streams_path}/{stream_id}");
- let stream_metadata = fs::metadata(stream_path).await.unwrap();
- assert!(stream_metadata.is_dir());
-}
diff --git a/core/integration/tests/streaming/stream.rs b/core/integration/tests/streaming/stream.rs
deleted file mode 100644
index 8d7b7cb8..00000000
--- a/core/integration/tests/streaming/stream.rs
+++ /dev/null
@@ -1,191 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::common::test_setup::TestSetup;
-use crate::streaming::create_messages;
-use ahash::AHashMap;
-use compio::fs;
-use iggy::prelude::*;
-use server::state::system::StreamState;
-use server::streaming::polling_consumer::PollingConsumer;
-use server::streaming::segments::IggyMessagesBatchMut;
-use server::streaming::streams::stream::Stream;
-
-#[compio::test]
-async fn should_persist_stream_with_topics_directory_and_info_file() {
- let setup = TestSetup::init().await;
- setup.create_streams_directory().await;
- let stream_ids = get_stream_ids();
- for stream_id in stream_ids {
- let name = format!("test-{stream_id}");
- let stream = Stream::create(
- stream_id,
- &name,
- setup.config.clone(),
- setup.storage.clone(),
- );
-
- stream.persist().await.unwrap();
-
- assert_persisted_stream(&stream.path, &setup.config.topic.path).await;
- }
-}
-
-#[compio::test]
-async fn should_load_existing_stream_from_disk() {
- let setup = TestSetup::init().await;
- setup.create_streams_directory().await;
- let stream_ids = get_stream_ids();
- for stream_id in stream_ids {
- let name = format!("test-{stream_id}");
- let stream = Stream::create(
- stream_id,
- &name,
- setup.config.clone(),
- setup.storage.clone(),
- );
- stream.persist().await.unwrap();
- assert_persisted_stream(&stream.path, &setup.config.topic.path).await;
-
- let mut loaded_stream = Stream::empty(
- stream_id,
- &name,
- setup.config.clone(),
- setup.storage.clone(),
- );
- let state = StreamState {
- id: stream_id,
- name: name.clone(),
- created_at: IggyTimestamp::now(),
- topics: AHashMap::new(),
- };
- loaded_stream.load(state).await.unwrap();
-
- assert_eq!(loaded_stream.stream_id, stream.stream_id);
- assert_eq!(loaded_stream.name, stream.name);
- assert_eq!(loaded_stream.path, stream.path);
- assert_eq!(loaded_stream.topics_path, stream.topics_path);
- }
-}
-
-#[compio::test]
-async fn should_delete_existing_stream_from_disk() {
- let setup = TestSetup::init().await;
- setup.create_streams_directory().await;
- let stream_ids = get_stream_ids();
- for stream_id in stream_ids {
- let name = format!("test-{stream_id}");
- let stream = Stream::create(
- stream_id,
- &name,
- setup.config.clone(),
- setup.storage.clone(),
- );
- stream.persist().await.unwrap();
- assert_persisted_stream(&stream.path, &setup.config.topic.path).await;
-
- stream.delete().await.unwrap();
-
- assert!(fs::metadata(&stream.path).await.is_err());
- }
-}
-
-#[compio::test]
-async fn should_purge_existing_stream_on_disk() {
- let setup = TestSetup::init().await;
- setup.create_streams_directory().await;
- let stream_ids = get_stream_ids();
- for stream_id in stream_ids {
- let name = format!("test-{stream_id}");
- let mut stream = Stream::create(
- stream_id,
- &name,
- setup.config.clone(),
- setup.storage.clone(),
- );
- stream.persist().await.unwrap();
- assert_persisted_stream(&stream.path, &setup.config.topic.path).await;
-
- let topic_id = 1;
- stream
- .create_topic(
- Some(topic_id),
- "test",
- 1,
- IggyExpiry::NeverExpire,
- Default::default(),
- MaxTopicSize::ServerDefault,
- 1,
- )
- .unwrap();
-
- let messages = create_messages();
- let messages_count = messages.len() as u32;
- let messages_size = messages
- .iter()
- .map(|msg| msg.get_size_bytes().as_bytes_u32())
- .sum::<u32>();
- let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- let topic = stream
- .get_topic(&Identifier::numeric(topic_id).unwrap())
- .unwrap();
- let partition_id = topic
- .calculate_partition_id(&Partitioning::partition_id(1))
- .unwrap();
- topic.append_messages(partition_id, batch).await.unwrap();
- let (_, loaded_messages) = topic
- .get_messages(
- PollingConsumer::Consumer(1, 1),
- 1,
- PollingStrategy::offset(0),
- 100,
- )
- .await
- .unwrap();
-
- assert_eq!(loaded_messages.count(), messages_count);
-
- // TODO: Fixme
- /*
- stream.purge().await.unwrap();
- let (metadata, loaded_messages) = topic
- .get_messages(
- PollingConsumer::Consumer(1, 1),
- 1,
- PollingStrategy::offset(0),
- 100,
- )
- .await
- .unwrap();
- assert_eq!(metadata.current_offset, 0);
- assert!(loaded_messages.is_empty());
- */
- }
-}
-
-async fn assert_persisted_stream(stream_path: &str, topics_directory: &str) {
- let stream_metadata = fs::metadata(stream_path).await.unwrap();
- assert!(stream_metadata.is_dir());
- let topics_path = format!("{stream_path}/{topics_directory}");
- let topics_metadata = fs::metadata(&topics_path).await.unwrap();
- assert!(topics_metadata.is_dir());
-}
-
-fn get_stream_ids() -> Vec<u32> {
- vec![1, 2, 3, 5, 10, 100, 1000, 99999]
-}
diff --git a/core/integration/tests/streaming/topic.rs b/core/integration/tests/streaming/topic.rs
deleted file mode 100644
index 26a02f96..00000000
--- a/core/integration/tests/streaming/topic.rs
+++ /dev/null
@@ -1,279 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::common::test_setup::TestSetup;
-use crate::streaming::create_messages;
-use ahash::AHashMap;
-use compio::fs;
-use iggy::prelude::*;
-use server::state::system::{PartitionState, TopicState};
-use server::streaming::polling_consumer::PollingConsumer;
-use server::streaming::segments::IggyMessagesBatchMut;
-use server::streaming::topics::topic::Topic;
-use std::default::Default;
-use std::sync::Arc;
-use std::sync::atomic::{AtomicU32, AtomicU64};
-
-#[compio::test]
-async fn should_persist_topics_with_partitions_directories_and_info_file() {
- let setup = TestSetup::init().await;
- let stream_id = 1;
- let partitions_count = 3;
- setup.create_topics_directory(stream_id).await;
- let topic_ids = get_topic_ids();
- for topic_id in topic_ids {
- let name = format!("test-{topic_id}");
- let created_topic_info = Topic::create(
- stream_id,
- topic_id,
- &name,
- partitions_count,
- setup.config.clone(),
- setup.storage.clone(),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- IggyExpiry::NeverExpire,
- CompressionAlgorithm::default(),
- MaxTopicSize::ServerDefault,
- 1,
- )
- .unwrap();
- let topic = created_topic_info.topic;
-
- topic.persist().await.unwrap();
-
- assert_persisted_topic(
- &topic.path,
- &setup
- .config
- .get_partitions_path(stream_id as usize, topic_id as usize),
- 3,
- )
- .await;
- }
-}
-
-#[compio::test]
-async fn should_load_existing_topic_from_disk() {
- let setup = TestSetup::init().await;
- let stream_id = 1;
- setup.create_topics_directory(stream_id).await;
- let partitions_count = 3;
- let topic_ids = get_topic_ids();
- for topic_id in topic_ids {
- let name = format!("test-{topic_id}");
- let created_topic_info = Topic::create(
- stream_id,
- topic_id,
- &name,
- partitions_count,
- setup.config.clone(),
- setup.storage.clone(),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- IggyExpiry::NeverExpire,
- CompressionAlgorithm::default(),
- MaxTopicSize::ServerDefault,
- 1,
- )
- .unwrap();
- let topic = created_topic_info.topic;
-
- topic.persist().await.unwrap();
- assert_persisted_topic(
- &topic.path,
- &setup
- .config
- .get_partitions_path(stream_id as usize, topic_id as usize),
- partitions_count,
- )
- .await;
-
- let created_at = IggyTimestamp::now();
- let created_topic_info = Topic::empty(
- stream_id,
- topic_id,
- &name,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- setup.config.clone(),
- setup.storage.clone(),
- );
- let mut loaded_topic = created_topic_info.topic;
- let topic_state = TopicState {
- id: topic_id,
- name,
- partitions: if partitions_count == 0 {
- AHashMap::new()
- } else {
- (1..=partitions_count)
- .map(|id| (id, PartitionState { id, created_at }))
- .collect()
- },
- consumer_groups: Default::default(),
- compression_algorithm: Default::default(),
- message_expiry: IggyExpiry::NeverExpire,
- max_topic_size: MaxTopicSize::ServerDefault,
- replication_factor: Some(1),
- created_at: Default::default(),
- };
- loaded_topic.load(topic_state).await.unwrap();
-
- assert_eq!(loaded_topic.stream_id, topic.stream_id);
- assert_eq!(loaded_topic.topic_id, topic.topic_id);
- assert_eq!(loaded_topic.name, topic.name);
- assert_eq!(
- loaded_topic.compression_algorithm,
- topic.compression_algorithm
- );
- assert_eq!(loaded_topic.path, topic.path);
- assert_eq!(loaded_topic.get_partitions().len() as u32, partitions_count);
- }
-}
-
-#[compio::test]
-async fn should_delete_existing_topic_from_disk() {
- let setup = TestSetup::init().await;
- let stream_id = 1;
- setup.create_topics_directory(stream_id).await;
- let partitions_count = 3;
- let topic_ids = get_topic_ids();
- for topic_id in topic_ids {
- let name = format!("test-{topic_id}");
- let created_topic_info = Topic::create(
- stream_id,
- topic_id,
- &name,
- partitions_count,
- setup.config.clone(),
- setup.storage.clone(),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- IggyExpiry::NeverExpire,
- CompressionAlgorithm::default(),
- MaxTopicSize::ServerDefault,
- 1,
- )
- .unwrap();
- let topic = created_topic_info.topic;
- topic.persist().await.unwrap();
- assert_persisted_topic(
- &topic.path,
- &setup
- .config
- .get_partitions_path(stream_id as usize, topic_id as usize),
- partitions_count,
- )
- .await;
-
- topic.delete().await.unwrap();
-
- assert!(fs::metadata(&topic.path).await.is_err());
- }
-}
-
-#[compio::test]
-async fn should_purge_existing_topic_on_disk() {
- let setup = TestSetup::init().await;
- let stream_id = 1;
- setup.create_topics_directory(stream_id).await;
- let partitions_count = 3;
- let topic_ids = get_topic_ids();
- for topic_id in topic_ids {
- let name = format!("test-{topic_id}");
- let created_topic_info = Topic::create(
- stream_id,
- topic_id,
- &name,
- partitions_count,
- setup.config.clone(),
- setup.storage.clone(),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- IggyExpiry::NeverExpire,
- CompressionAlgorithm::default(),
- MaxTopicSize::ServerDefault,
- 1,
- )
- .unwrap();
- let topic = created_topic_info.topic;
- topic.persist().await.unwrap();
- assert_persisted_topic(
- &topic.path,
- &setup
- .config
- .get_partitions_path(stream_id as usize, topic_id as usize),
- partitions_count,
- )
- .await;
-
- let messages = create_messages();
- let messages_count = messages.len() as u32;
- let batch_size = messages
- .iter()
- .map(|msg| msg.get_size_bytes().as_bytes_u32())
- .sum::<u32>();
- let batch = IggyMessagesBatchMut::from_messages(&messages, batch_size);
- let partition_id = topic
- .calculate_partition_id(&Partitioning::partition_id(1))
- .unwrap();
- topic.append_messages(partition_id, batch).await.unwrap();
- let (_, loaded_messages) = topic
- .get_messages(
- PollingConsumer::Consumer(1, 1),
- 1,
- PollingStrategy::offset(0),
- 100,
- )
- .await
- .unwrap();
- assert_eq!(loaded_messages.count(), messages_count);
-
- topic.purge().await.unwrap();
- let (metadata, loaded_messages) = topic
- .get_messages(
- PollingConsumer::Consumer(1, 1),
- 1,
- PollingStrategy::offset(0),
- 100,
- )
- .await
- .unwrap();
- assert_eq!(metadata.current_offset, 0);
- assert!(loaded_messages.is_empty());
- }
-}
-
-async fn assert_persisted_topic(topic_path: &str, partitions_path: &str, partitions_count: u32) {
- let topic_metadata = fs::metadata(topic_path).await.unwrap();
- assert!(topic_metadata.is_dir());
- for partition_id in 1..=partitions_count {
- let partition_path = format!("{partitions_path}/{partition_id}");
- let partition_metadata = fs::metadata(&partition_path).await.unwrap();
- assert!(partition_metadata.is_dir());
- }
-}
-
-fn get_topic_ids() -> Vec<u32> {
- vec![1, 2, 3, 5, 10, 100, 1000, 99999]
-}
diff --git a/core/integration/tests/streaming/topic_messages.rs b/core/integration/tests/streaming/topic_messages.rs
deleted file mode 100644
index 2e46280a..00000000
--- a/core/integration/tests/streaming/topic_messages.rs
+++ /dev/null
@@ -1,245 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::common::test_setup::TestSetup;
-use bytes::Bytes;
-use iggy::prelude::*;
-use iggy_common::locking::IggyRwLockFn;
-use server::configs::system::SystemConfig;
-use server::streaming::polling_consumer::PollingConsumer;
-use server::streaming::segments::IggyMessagesBatchMut;
-use server::streaming::topics::topic::Topic;
-use server::streaming::utils::hash;
-use std::collections::HashMap;
-use std::str::from_utf8;
-use std::sync::Arc;
-use std::sync::atomic::{AtomicU32, AtomicU64};
-
-#[compio::test]
-async fn assert_polling_messages() {
- let messages_count = 1000;
- let payload_size_bytes = 1000;
- let config = SystemConfig {
- ..Default::default()
- };
- let setup = TestSetup::init_with_config(config).await;
- let topic = init_topic(&setup, 1).await;
- let partition_id = 1;
- let partitioning = Partitioning::partition_id(partition_id);
- let messages = (0..messages_count)
- .map(|id| {
- IggyMessage::builder()
- .id(id as u128)
- .payload(Bytes::from(format!(
- "{}:{}",
- id + 1,
- create_payload(payload_size_bytes)
- )))
- .build()
- .expect("Failed to create message")
- })
- .collect::<Vec<_>>();
- let mut sent_messages = Vec::new();
- for message in &messages {
- sent_messages.push(message);
- }
- let batch_size = messages
- .iter()
- .map(|m| m.get_size_bytes())
- .sum::<IggyByteSize>();
- let batch = IggyMessagesBatchMut::from_messages(&messages, batch_size.as_bytes_u32());
- let partition_id = topic.calculate_partition_id(&partitioning).unwrap();
- topic.append_messages(partition_id, batch).await.unwrap();
-
- let consumer = PollingConsumer::Consumer(1, partition_id);
- let (_, polled_messages) = topic
- .get_messages(
- consumer,
- partition_id,
- PollingStrategy::offset(0),
- messages_count,
- )
- .await
- .unwrap();
- assert_eq!(topic.get_messages_count(), messages_count as u64);
- assert_eq!(polled_messages.count(), messages_count);
- let mut cnt = 0;
- for batch in polled_messages.iter() {
- for polled_message in batch.iter() {
- let sent_message = sent_messages.get(cnt).unwrap();
- assert_eq!(sent_message.payload, polled_message.payload());
- let polled_payload_str = from_utf8(polled_message.payload()).unwrap();
- assert!(polled_payload_str.starts_with(&format!("{}:", cnt + 1)));
- cnt += 1;
- }
- }
-}
-
-#[compio::test]
-async fn given_key_none_messages_should_be_appended_to_the_next_partition_using_round_robin() {
- let setup = TestSetup::init().await;
- let partitions_count = 3;
- let messages_per_partition_count = 10;
- let topic = init_topic(&setup, partitions_count).await;
- let partitioning = Partitioning::balanced();
- for i in 1..=partitions_count * messages_per_partition_count {
- let payload = get_payload(i);
- let batch_size = IggyByteSize::from(16 + 4 + payload.len() as u64);
- let messages = IggyMessagesBatchMut::from_messages(
- &[IggyMessage::builder()
- .id(i as u128)
- .payload(Bytes::from(payload.to_owned()))
- .build()
- .expect("Failed to create message")],
- batch_size.as_bytes_u32(),
- );
- let partition_id = topic.calculate_partition_id(&partitioning).unwrap();
- topic.append_messages(partition_id, messages).await.unwrap();
- }
- for i in 1..=partitions_count {
- assert_messages(&topic, i, messages_per_partition_count).await;
- }
-}
-
-#[compio::test]
-async fn given_key_partition_id_messages_should_be_appended_to_the_chosen_partition() {
- let setup = TestSetup::init().await;
- let partition_id = 1;
- let partitions_count = 3;
- let messages_per_partition_count = 10;
- let topic = init_topic(&setup, partitions_count).await;
- let partitioning = Partitioning::partition_id(partition_id);
- for i in 1..=partitions_count * messages_per_partition_count {
- let payload = get_payload(i);
- let batch_size = IggyByteSize::from(16 + 4 + payload.len() as u64);
- let messages = IggyMessagesBatchMut::from_messages(
- &[IggyMessage::builder()
- .id(i as u128)
- .payload(Bytes::from(payload.to_owned()))
- .build()
- .expect("Failed to create message")],
- batch_size.as_bytes_u32(),
- );
- let partition_id = topic.calculate_partition_id(&partitioning).unwrap();
- topic.append_messages(partition_id, messages).await.unwrap();
- }
-
- for i in 1..=partitions_count {
- if i == partition_id {
- assert_messages(&topic, i, messages_per_partition_count * partitions_count).await;
- continue;
- }
- assert_messages(&topic, i, 0).await;
- }
-}
-
-#[compio::test]
-async fn given_key_messages_key_messages_should_be_appended_to_the_calculated_partition() {
- let setup = TestSetup::init().await;
- let partitions_count = 3;
- let messages_count = 10;
- let topic = init_topic(&setup, partitions_count).await;
- for entity_id in 1..=partitions_count * messages_count {
- let payload = get_payload(entity_id);
- let partitioning = Partitioning::messages_key_u32(entity_id);
- let batch_size = IggyByteSize::from(16 + 4 + payload.len() as u64);
- let messages = IggyMessagesBatchMut::from_messages(
- &[IggyMessage::builder()
- .id(entity_id as u128)
- .payload(Bytes::from(payload.to_owned()))
- .build()
- .expect("Failed to create message")],
- batch_size.as_bytes_u32(),
- );
- let partition_id = topic.calculate_partition_id(&partitioning).unwrap();
- topic.append_messages(partition_id, messages).await.unwrap();
- }
-
- let mut messages_count_per_partition = HashMap::new();
- for entity_id in 1..=partitions_count * messages_count {
- let key = Partitioning::messages_key_u32(entity_id);
- let hash = hash::calculate_32(&key.value);
- let mut partition_id = hash % partitions_count;
- if partition_id == 0 {
- partition_id = partitions_count;
- }
-
- messages_count_per_partition
- .entry(partition_id)
- .and_modify(|count| *count += 1)
- .or_insert(1);
- }
-
- for (partition_id, expected_messages) in messages_count_per_partition {
- assert_messages(&topic, partition_id, expected_messages).await;
- }
-}
-
-fn get_payload(id: u32) -> String {
- format!("message-{id}")
-}
-
-async fn assert_messages(topic: &Topic, partition_id: u32, expected_messages: u32) {
- let consumer = PollingConsumer::Consumer(0, partition_id);
- let (_, polled_messages) = topic
- .get_messages(consumer, partition_id, PollingStrategy::offset(0), 1000)
- .await
- .unwrap();
- assert_eq!(polled_messages.count(), expected_messages);
-}
-
-async fn init_topic(setup: &TestSetup, partitions_count: u32) -> Topic {
- let stream_id = 1;
- setup.create_topics_directory(stream_id).await;
- let id = 2;
- let name = "test";
- let created_topic_info = Topic::create(
- stream_id,
- id,
- name,
- partitions_count,
- setup.config.clone(),
- setup.storage.clone(),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU32::new(0)),
- IggyExpiry::NeverExpire,
- Default::default(),
- MaxTopicSize::ServerDefault,
- 1,
- )
- .unwrap();
- let topic = created_topic_info.topic;
- topic.persist().await.unwrap();
- for partition in topic.get_partitions() {
- let mut partition = partition.write().await;
- partition.persist().await.unwrap();
- partition.open().await.unwrap();
- }
- topic
-}
-
-fn create_payload(size: u32) -> String {
- let mut payload = String::with_capacity(size as usize);
- for i in 0..size {
- let char = (i % 26 + 97) as u8 as char;
- payload.push(char);
- }
-
- payload
-}
diff --git a/core/server/src/binary/handlers/messages/poll_messages_handler.rs b/core/server/src/binary/handlers/messages/poll_messages_handler.rs
index 37aaac20..1a835efc 100644
--- a/core/server/src/binary/handlers/messages/poll_messages_handler.rs
+++ b/core/server/src/binary/handlers/messages/poll_messages_handler.rs
@@ -81,8 +81,8 @@ impl ServerCommandHandler for PollMessages {
.poll_messages(
client_id,
user_id,
- &stream_id,
- &topic_id,
+ stream_id,
+ topic_id,
consumer,
partition_id,
args,
diff --git a/core/server/src/binary/handlers/messages/send_messages_handler.rs b/core/server/src/binary/handlers/messages/send_messages_handler.rs
index 6b998c31..f8afd67a 100644
--- a/core/server/src/binary/handlers/messages/send_messages_handler.rs
+++ b/core/server/src/binary/handlers/messages/send_messages_handler.rs
@@ -114,8 +114,8 @@ impl ServerCommandHandler for SendMessages {
shard
.append_messages(
user_id,
- &self.stream_id,
- &self.topic_id,
+ self.stream_id,
+ self.topic_id,
&self.partitioning,
batch,
)
diff --git a/core/server/src/configs/system.rs b/core/server/src/configs/system.rs
index f119b3e3..add75b31 100644
--- a/core/server/src/configs/system.rs
+++ b/core/server/src/configs/system.rs
@@ -23,6 +23,7 @@ use crate::slab::topics;
use super::cache_indexes::CacheIndexesConfig;
use super::sharding::ShardingConfig;
use iggy_common::IggyByteSize;
+use iggy_common::IggyError;
use iggy_common::IggyExpiry;
use iggy_common::MaxTopicSize;
use iggy_common::{CompressionAlgorithm, IggyDuration};
@@ -292,4 +293,30 @@ impl SystemConfig {
let path = self.get_segment_path(stream_id, topic_id, partition_id, start_offset);
format!("{path}.{INDEX_EXTENSION}")
}
+
+ pub fn resolve_max_topic_size(
+ &self,
+ max_topic_size: MaxTopicSize,
+ ) -> Result<MaxTopicSize, IggyError> {
+ match max_topic_size {
+ MaxTopicSize::ServerDefault => Ok(self.topic.max_size),
+ _ => {
+ if max_topic_size.as_bytes_u64() < self.segment.size.as_bytes_u64() {
+ Err(IggyError::InvalidTopicSize(
+ max_topic_size,
+ self.segment.size,
+ ))
+ } else {
+ Ok(max_topic_size)
+ }
+ }
+ }
+ }
+
+ pub fn resolve_message_expiry(&self, message_expiry: IggyExpiry) -> IggyExpiry {
+ match message_expiry {
+ IggyExpiry::ServerDefault => self.segment.message_expiry,
+ _ => message_expiry,
+ }
+ }
}
diff --git a/core/server/src/http/http_shard_wrapper.rs b/core/server/src/http/http_shard_wrapper.rs
index e6c29276..4ac0cddf 100644
--- a/core/server/src/http/http_shard_wrapper.rs
+++ b/core/server/src/http/http_shard_wrapper.rs
@@ -255,8 +255,8 @@ impl HttpSafeShard {
&self,
client_id: u32,
user_id: u32,
- stream_id: &Identifier,
- topic_id: &Identifier,
+ stream_id: Identifier,
+ topic_id: Identifier,
consumer: Consumer,
maybe_partition_id: Option<u32>,
args: PollingArgs,
@@ -277,8 +277,8 @@ impl HttpSafeShard {
pub async fn append_messages(
&self,
user_id: u32,
- stream_id: &Identifier,
- topic_id: &Identifier,
+ stream_id: Identifier,
+ topic_id: Identifier,
partitioning: &Partitioning,
batch: IggyMessagesBatchMut,
) -> Result<(), IggyError> {
diff --git a/core/server/src/http/messages.rs b/core/server/src/http/messages.rs
index 86dccf74..1acc4d02 100644
--- a/core/server/src/http/messages.rs
+++ b/core/server/src/http/messages.rs
@@ -68,8 +68,8 @@ async fn poll_messages(
let poll_future = SendWrapper::new(state.shard.poll_messages(
session.client_id,
session.get_user_id(),
- &query.0.stream_id,
- &query.0.topic_id,
+ query.0.stream_id,
+ query.0.topic_id,
consumer,
query.0.partition_id,
PollingArgs::new(query.0.strategy, query.0.count, query.0.auto_commit),
@@ -106,8 +106,8 @@ async fn send_messages(
let append_future = SendWrapper::new(state.shard.append_messages(
identity.user_id,
- &command_stream_id,
- &command_topic_id,
+ command_stream_id,
+ command_topic_id,
&partitioning,
batch,
));
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 8794f665..d4f57a89 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -60,7 +60,7 @@ use crate::{
http::http_server,
io::fs_utils,
shard::{
- namespace::IggyNamespace,
+ namespace::{IggyFullNamespace, IggyNamespace},
task_registry::TaskRegistry,
tasks::messages::spawn_shard_message_task,
transmission::{
@@ -85,6 +85,7 @@ use crate::{
session::Session,
storage::SystemStorage,
streams, topics,
+ traits::MainOps,
users::{permissioner::Permissioner, user::User},
utils::ptr::EternalPtr,
},
@@ -354,79 +355,13 @@ impl IggyShard {
let partition_id = request.partition_id;
match request.payload {
ShardRequestPayload::SendMessages { batch } => {
- let mut batch = self.maybe_encrypt_messages(batch)?;
+ let ns = IggyFullNamespace::new(stream_id, topic_id, partition_id);
+ let batch = self.maybe_encrypt_messages(batch)?;
let messages_count = batch.count();
-
- let current_offset = self.streams2.with_partition_by_id(
- &stream_id,
- &topic_id,
- partition_id,
- partitions::helpers::calculate_current_offset(),
- );
-
self.streams2
- .with_partition_by_id_async(
- &stream_id,
- &topic_id,
- partition_id,
- partitions::helpers::deduplicate_messages(current_offset, &mut batch),
- )
- .await;
-
- let (journal_messages_count, journal_size) =
- self.streams2.with_partition_by_id_mut(
- &stream_id,
- &topic_id,
- partition_id,
- partitions::helpers::append_to_journal(self.id, current_offset, batch),
- )?;
-
- let unsaved_messages_count_exceeded = journal_messages_count
- >= self.config.system.partition.messages_required_to_save;
- let unsaved_messages_size_exceeded = journal_size
- >= self
- .config
- .system
- .partition
- .size_of_messages_required_to_save
- .as_bytes_u64() as u32;
-
- let is_full = self.streams2.with_partition_by_id(
- &stream_id,
- &topic_id,
- partition_id,
- partitions::helpers::is_segment_full(),
- );
-
- // Try committing the journal
- if is_full || unsaved_messages_count_exceeded || unsaved_messages_size_exceeded {
- self.streams2
- .persist_messages(
- self.id,
- &stream_id,
- &topic_id,
- partition_id,
- unsaved_messages_count_exceeded,
- unsaved_messages_size_exceeded,
- journal_messages_count,
- journal_size,
- &self.config.system,
- )
- .await?;
-
- if is_full {
- self.streams2
- .handle_full_segment(
- self.id,
- &stream_id,
- &topic_id,
- partition_id,
- &self.config.system,
- )
- .await?;
- self.metrics.increment_messages(messages_count as u64);
- }
- }
+ .append_messages(self.id, &self.config.system, &ns, batch)
+ .await?;
+ self.metrics.increment_messages(messages_count as u64);
Ok(ShardResponse::SendMessages)
}
ShardRequestPayload::PollMessages { args, consumer } => {
diff --git a/core/server/src/shard/namespace.rs b/core/server/src/shard/namespace.rs
index f3a8f2c3..5c66e8db 100644
--- a/core/server/src/shard/namespace.rs
+++ b/core/server/src/shard/namespace.rs
@@ -17,8 +17,11 @@
*/
use hash32::{Hasher, Murmur3Hasher};
+use iggy_common::Identifier;
use std::hash::Hasher as _;
+use crate::slab::partitions;
+
// Packed namespace layout (works only on 64bit platforms, but we won't support 32bit anyway)
// +----------------+----------------+----------------+----------------+
// | stream_id | topic_id | partition_id | unused |
@@ -54,6 +57,38 @@ pub const PARTITION_MASK: u64 = (1u64 << PARTITION_BITS) - 1;
pub const TOPIC_MASK: u64 = (1u64 << TOPIC_BITS) - 1;
pub const STREAM_MASK: u64 = (1u64 << STREAM_BITS) - 1;
+pub struct IggyFullNamespace {
+ stream: Identifier,
+ topic: Identifier,
+ partition: partitions::ContainerId,
+}
+
+impl IggyFullNamespace {
+ pub fn new(stream: Identifier, topic: Identifier, partition: partitions::ContainerId) -> Self {
+ Self {
+ stream,
+ topic,
+ partition,
+ }
+ }
+
+ pub fn stream_id(&self) -> &Identifier {
+ &self.stream
+ }
+
+ pub fn topic_id(&self) -> &Identifier {
+ &self.topic
+ }
+
+ pub fn partition_id(&self) -> partitions::ContainerId {
+ self.partition
+ }
+
+ pub fn decompose(self) -> (Identifier, Identifier, partitions::ContainerId) {
+ (self.stream, self.topic, self.partition)
+ }
+}
+
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct IggyNamespace(u64);
diff --git a/core/server/src/shard/system/messages.rs b/core/server/src/shard/system/messages.rs
index 1a57f4a1..6acf72d1 100644
--- a/core/server/src/shard/system/messages.rs
+++ b/core/server/src/shard/system/messages.rs
@@ -21,7 +21,7 @@ use std::sync::atomic::Ordering;
use super::COMPONENT;
use crate::binary::handlers::messages::poll_messages_handler::IggyPollMetadata;
use crate::shard::IggyShard;
-use crate::shard::namespace::IggyNamespace;
+use crate::shard::namespace::{IggyFullNamespace, IggyNamespace};
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{
ShardMessage, ShardRequest, ShardRequestPayload, ShardSendRequestResult,
@@ -30,6 +30,7 @@ use crate::shard_trace;
use crate::streaming::polling_consumer::PollingConsumer;
use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut, IggyMessagesBatchSet};
use crate::streaming::session::Session;
+use crate::streaming::traits::MainOps;
use crate::streaming::utils::{PooledBuffer, hash};
use crate::streaming::{partitions, streams, topics};
use error_set::ErrContext;
@@ -44,8 +45,8 @@ impl IggyShard {
pub async fn append_messages(
&self,
user_id: u32,
- stream_id: &Identifier,
- topic_id: &Identifier,
+ stream_id: Identifier,
+ topic_id: Identifier,
partitioning: &Partitioning,
batch: IggyMessagesBatchMut,
) -> Result<(), IggyError> {
@@ -72,11 +73,11 @@ impl IggyShard {
let numeric_stream_id = self
.streams2
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
+ .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());
+ .with_topic_by_id(&stream_id, &topic_id, topics::helpers::get_topic_id());
// Validate permissions for given user on stream and topic.
self.permissioner
@@ -97,8 +98,8 @@ impl IggyShard {
let partition_id =
self.streams2
.with_topic_by_id(
- stream_id,
- topic_id,
+ &stream_id,
+ &topic_id,
|(root, auxilary, ..)| match partitioning.kind {
PartitioningKind::Balanced => {
let upperbound = root.partitions().len();
@@ -121,7 +122,7 @@ impl IggyShard {
)?;
let namespace = IggyNamespace::new(numeric_stream_id, numeric_topic_id, partition_id);
- let payload = ShardRequestPayload::SendMessages { batch: batch };
+ let payload = ShardRequestPayload::SendMessages { batch };
let request = ShardRequest::new(stream_id.clone(), topic_id.clone(), partition_id, payload);
let message = ShardMessage::Request(request);
match self
@@ -130,87 +131,21 @@ impl IggyShard {
{
ShardSendRequestResult::Recoil(message) => {
if let ShardMessage::Request(ShardRequest {
+ stream_id,
+ topic_id,
partition_id,
payload,
- ..
}) = message
&& let ShardRequestPayload::SendMessages { batch } = payload
{
+ let ns = IggyFullNamespace::new(stream_id, topic_id, partition_id);
// Encrypt messages if encryptor is enabled in configuration.
- let mut batch = self.maybe_encrypt_messages(batch)?;
+ let batch = self.maybe_encrypt_messages(batch)?;
let messages_count = batch.count();
-
- let current_offset = self.streams2.with_partition_by_id(
- stream_id,
- topic_id,
- partition_id,
- partitions::helpers::calculate_current_offset(),
- );
-
self.streams2
- .with_partition_by_id_async(
- stream_id,
- topic_id,
- partition_id,
- partitions::helpers::deduplicate_messages(current_offset, &mut batch),
- )
- .await;
-
- let (journal_messages_count, journal_size) =
- self.streams2.with_partition_by_id_mut(
- stream_id,
- topic_id,
- partition_id,
- partitions::helpers::append_to_journal(self.id, current_offset, batch),
- )?;
-
- let unsaved_messages_count_exceeded = journal_messages_count
- >= self.config.system.partition.messages_required_to_save;
- let unsaved_messages_size_exceeded = journal_size
- >= self
- .config
- .system
- .partition
- .size_of_messages_required_to_save
- .as_bytes_u64() as u32;
-
- let is_full = self.streams2.with_partition_by_id(
- stream_id,
- topic_id,
- partition_id,
- partitions::helpers::is_segment_full(),
- );
-
- // Try committing the journal
- if is_full || unsaved_messages_count_exceeded || unsaved_messages_size_exceeded
- {
- self.streams2
- .persist_messages(
- self.id,
- stream_id,
- topic_id,
- partition_id,
- unsaved_messages_count_exceeded,
- unsaved_messages_size_exceeded,
- journal_messages_count,
- journal_size,
- &self.config.system,
- )
- .await?;
-
- if is_full {
- self.streams2
- .handle_full_segment(
- self.id,
- stream_id,
- topic_id,
- partition_id,
- &self.config.system,
- )
- .await?;
- self.metrics.increment_messages(messages_count as u64);
- }
- }
+ .append_messages(self.id, &self.config.system, &ns, batch)
+ .await?;
+ self.metrics.increment_messages(messages_count as u64);
Ok(())
} else {
unreachable!(
@@ -234,18 +169,18 @@ impl IggyShard {
&self,
client_id: u32,
user_id: u32,
- stream_id: &Identifier,
- topic_id: &Identifier,
+ stream_id: Identifier,
+ topic_id: Identifier,
consumer: Consumer,
maybe_partition_id: Option<u32>,
args: PollingArgs,
) -> Result<(IggyPollMetadata, IggyMessagesBatchSet), IggyError> {
let numeric_stream_id = self
.streams2
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
+ .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());
+ .with_topic_by_id(&stream_id, &topic_id, topics::helpers::get_topic_id());
self.permissioner
.borrow()
@@ -259,8 +194,8 @@ impl IggyShard {
// Resolve partition ID
let Some((consumer, partition_id)) = self.resolve_consumer_with_partition_id(
- stream_id,
- topic_id,
+ &stream_id,
+ &topic_id,
&consumer,
client_id,
maybe_partition_id,
@@ -271,7 +206,7 @@ impl IggyShard {
let has_partition = self
.streams2
- .with_topic_by_id(stream_id, topic_id, |(root, ..)| {
+ .with_topic_by_id(&stream_id, &topic_id, |(root, ..)| {
root.partitions().exists(partition_id)
});
if !has_partition {
@@ -282,8 +217,8 @@ impl IggyShard {
}
let current_offset = self.streams2.with_partition_by_id(
- stream_id,
- topic_id,
+ &stream_id,
+ &topic_id,
partition_id,
|(_, _, _, offset, ..)| offset.load(Ordering::Relaxed),
);
@@ -312,151 +247,14 @@ impl IggyShard {
}) = message
&& let ShardRequestPayload::PollMessages { consumer, args } = payload
{
- let metadata = IggyPollMetadata::new(partition_id as u32, current_offset);
- let count = args.count;
- let strategy = args.strategy;
- let value = strategy.value;
- let batches = match strategy.kind {
- PollingKind::Offset => {
- let offset = value;
- // We have to remember to keep the invariant from the if that is on line 496.
- // Alternatively a better design would be to get rid of that if and move the validations here.
- let batches = self
- .streams2
- .get_messages_by_offset(
- stream_id,
- topic_id,
- partition_id,
- offset,
- count,
- )
- .await?;
- Ok(batches)
- }
- PollingKind::Timestamp => {
- let timestamp = IggyTimestamp::from(value);
- let timestamp_ts = timestamp.as_micros();
- trace!(
- "Getting {count} messages by timestamp: {} for partition: {}...",
- timestamp_ts, partition_id
- );
-
- let batches = self
- .streams2
- .get_messages_by_timestamp(
- stream_id,
- topic_id,
- partition_id,
- timestamp_ts,
- count,
- )
- .await?;
- Ok(batches)
- }
- PollingKind::First => {
- let first_offset = self.streams2.with_partition_by_id(
- stream_id,
- topic_id,
- partition_id,
- |(_, _, _, _, _, _, log)| {
- log.segments()
- .first()
- .map(|segment| segment.start_offset)
- .unwrap_or(0)
- },
- );
-
- let batches = self
- .streams2
- .get_messages_by_offset(
- stream_id,
- topic_id,
- partition_id,
- first_offset,
- count,
- )
- .await?;
- Ok(batches)
- }
- PollingKind::Last => {
- let (start_offset, actual_count) = self.streams2.with_partition_by_id(
- stream_id,
- topic_id,
- partition_id,
- |(_, _, _, offset, _, _, _)| {
- let current_offset = offset.load(Ordering::Relaxed);
- let mut requested_count = 0;
- if requested_count > current_offset + 1 {
- requested_count = current_offset + 1
- }
- let start_offset = 1 + current_offset - requested_count;
- (start_offset, requested_count as u32)
- },
- );
-
- let batches = self
- .streams2
- .get_messages_by_offset(
- stream_id,
- topic_id,
- partition_id,
- start_offset,
- actual_count,
- )
- .await?;
- Ok(batches)
- }
- PollingKind::Next => {
- let (consumer_offset, consumer_id) = match consumer {
- PollingConsumer::Consumer(consumer_id, _) => (
- self.streams2
- .with_partition_by_id(
- stream_id,
- topic_id,
- partition_id,
- partitions::helpers::get_consumer_offset(consumer_id),
- )
- .map(|c_offset| c_offset.stored_offset),
- consumer_id,
- ),
- PollingConsumer::ConsumerGroup(cg_id, _) => (
- self.streams2
- .with_partition_by_id(
- stream_id,
- topic_id,
- partition_id,
- partitions::helpers::get_consumer_group_member_offset(
- cg_id,
- ),
- )
- .map(|cg_offset| cg_offset.stored_offset),
- cg_id,
- ),
- };
-
- let Some(consumer_offset) = consumer_offset else {
- return Err(IggyError::ConsumerOffsetNotFound(consumer_id));
- };
- 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)
- }
- }?;
-
- if args.auto_commit && !batches.is_empty() {
+ let ns = IggyFullNamespace::new(stream_id, topic_id, partition_id);
+ let auto_commit = args.auto_commit;
+ let (metadata, batches) =
+ self.streams2.poll_messages(&ns, consumer, args).await?;
+ let stream_id = ns.stream_id();
+ let topic_id = ns.topic_id();
+
+ if auto_commit && !batches.is_empty() {
let offset = batches
.last_offset()
.expect("Batch set should have at least one batch");
@@ -467,8 +265,8 @@ impl IggyShard {
match consumer {
PollingConsumer::Consumer(consumer_id, _) => {
self.streams2.with_partition_by_id(
- stream_id,
- topic_id,
+ &stream_id,
+ &topic_id,
partition_id,
partitions::helpers::store_consumer_offset(
consumer_id,
@@ -481,8 +279,8 @@ impl IggyShard {
);
self.streams2
.with_partition_by_id_async(
- stream_id,
- topic_id,
+ &stream_id,
+ &topic_id,
partition_id,
partitions::helpers::persist_consumer_offset_to_disk(
self.id,
@@ -493,8 +291,8 @@ impl IggyShard {
}
PollingConsumer::ConsumerGroup(cg_id, _) => {
self.streams2.with_partition_by_id(
- stream_id,
- topic_id,
+ &stream_id,
+ &topic_id,
partition_id,
partitions::helpers::store_consumer_group_member_offset(
cg_id,
@@ -506,8 +304,8 @@ impl IggyShard {
),
);
self.streams2.with_partition_by_id_async(
- stream_id,
- topic_id,
+ &stream_id,
+ &topic_id,
partition_id,
partitions::helpers::persist_consumer_group_member_offset_to_disk(
self.id,
diff --git a/core/server/src/shard/system/partitions.rs b/core/server/src/shard/system/partitions.rs
index 7b1eed84..432eb0fd 100644
--- a/core/server/src/shard/system/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -96,11 +96,13 @@ impl IggyShard {
let parent_stats =
self.streams2
.with_topic_by_id(stream_id, topic_id, topics::helpers::get_stats());
- let partitions = self.create_and_insert_partitions_mem(
+ let partitions = partition2::create_and_insert_partitions_mem(
+ &self.streams2,
stream_id,
topic_id,
parent_stats,
partitions_count,
+ &self.config.system,
);
let stats = partitions.first().map(|p| p.stats());
if let Some(stats) = stats {
@@ -133,57 +135,6 @@ impl IggyShard {
Ok(partitions)
}
- fn create_and_insert_partitions_mem(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- parent_stats: Arc<TopicStats>,
- partitions_count: u32,
- ) -> Vec<partition2::Partition> {
- let range = 0..partitions_count as usize;
- let created_at = IggyTimestamp::now();
- range
- .map(|_| {
- // Areczkuuuu.
- let stats = Arc::new(PartitionStats::new(parent_stats.clone()));
- let should_increment_offset = false;
- let deduplicator = create_message_deduplicator(&self.config.system);
- let offset = Arc::new(AtomicU64::new(0));
- let consumer_offset = Arc::new(partition2::ConsumerOffsets::with_capacity(2137));
- let consumer_group_offset =
- Arc::new(partition2::ConsumerGroupOffsets::with_capacity(2137));
- let log = Default::default();
-
- let mut partition = partition2::Partition::new(
- created_at,
- should_increment_offset,
- stats,
- deduplicator,
- offset,
- consumer_offset,
- consumer_group_offset,
- log,
- );
- let id = self.insert_partition_mem(stream_id, topic_id, partition.clone());
- partition.update_id(id);
- partition
- })
- .collect()
- }
-
- fn insert_partition_mem(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- partition: partition2::Partition,
- ) -> usize {
- self.streams2.with_partitions_mut(
- stream_id,
- topic_id,
- partitions::helpers::insert_partition(partition),
- )
- }
-
async fn init_log(
&self,
stream_id: &Identifier,
@@ -263,7 +214,11 @@ impl IggyShard {
.with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
for partition in partitions {
let actual_id = partition.id();
- let id = self.insert_partition_mem(stream_id, topic_id, partition);
+ let id = self.streams2.with_partitions_mut(
+ stream_id,
+ topic_id,
+ partitions::helpers::insert_partition(partition),
+ );
assert_eq!(
id, actual_id,
"create_partitions_bypass_auth: partition mismatch ID, wrong creation order ?!"
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index 470d5d4f..fb8f445b 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -48,21 +48,12 @@ impl IggyShard {
if exists {
return Err(IggyError::StreamNameAlreadyExists(name));
}
- let stream = self.create_and_insert_stream_mem(name);
+ let stream = stream2::create_and_insert_stream_mem(&self.streams2, name);
self.metrics.increment_streams(1);
create_stream_file_hierarchy(self.id, stream.id(), &self.config.system).await?;
Ok(stream)
}
- fn create_and_insert_stream_mem(&self, name: String) -> stream2::Stream {
- let now = IggyTimestamp::now();
- let stats = Arc::new(Default::default());
- let mut stream = stream2::Stream::new(name, stats, now);
- let id = self.streams2.insert(stream.clone());
- stream.update_id(id);
- stream
- }
-
pub fn create_stream2_bypass_auth(&self, stream: stream2::Stream) -> usize {
self.streams2.insert(stream)
}
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index 642f5430..a9cf7d9f 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -23,7 +23,7 @@ use crate::slab::traits_ext::{EntityMarker, InsertCell};
use crate::streaming::session::Session;
use crate::streaming::stats::stats::{StreamStats, TopicStats};
use crate::streaming::topics::storage2::{create_topic_file_hierarchy, delete_topic_from_disk};
-use crate::streaming::topics::topic2::{self, resolve_max_topic_size, resolve_message_expiry};
+use crate::streaming::topics::topic2::{self};
use crate::streaming::{partitions, streams, topics};
use error_set::ErrContext;
use iggy_common::{
@@ -72,10 +72,11 @@ impl IggyShard {
let parent_stats = self
.streams2
.with_stream_by_id(stream_id, |(_, stats)| stats.clone());
- let message_expiry = resolve_message_expiry(message_expiry, config);
+ let message_expiry = config.resolve_message_expiry(message_expiry);
shard_info!(self.id, "Topic message expiry: {}", message_expiry);
- let max_topic_size = resolve_max_topic_size(max_topic_size, config)?;
- let topic = self.create_and_insert_topics_mem(
+ let max_topic_size = config.resolve_max_topic_size(max_topic_size)?;
+ let topic = topic2::create_and_insert_topics_mem(
+ &self.streams2,
stream_id,
name,
replication_factor.unwrap_or(1),
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index 0c84f7a3..0343231f 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -1,19 +1,20 @@
use crate::{
+ binary::handlers::messages::poll_messages_handler::IggyPollMetadata,
configs::{cache_indexes::CacheIndexesConfig, system::SystemConfig},
+ shard::{namespace::IggyFullNamespace, system::messages::PollingArgs},
+ shard_info,
slab::{
- Keyed,
- consumer_groups::ConsumerGroups,
- helpers,
- partitions::{self, Partitions},
- topics::Topics,
- traits_ext::{
+ consumer_groups::ConsumerGroups, helpers, partitions::{self, Partitions}, topics::Topics, traits_ext::{
ComponentsById, DeleteCell, EntityComponentSystem, EntityComponentSystemMutCell,
InsertCell, InteriorMutability, IntoComponents,
- },
+ }, Keyed
},
streaming::{
- partitions::partition2::{PartitionRef, PartitionRefMut},
- segments::{IggyMessagesBatchSet, Segment2, storage::create_segment_storage},
+ partitions::{journal::Journal, partition2::{PartitionRef, PartitionRefMut}},
+ polling_consumer::PollingConsumer,
+ segments::{
+ storage::create_segment_storage, IggyMessagesBatchMut, IggyMessagesBatchSet, Segment2
+ },
stats::stats::StreamStats,
streams::{
self,
@@ -24,12 +25,17 @@ use crate::{
consumer_group2::{ConsumerGroupRef, ConsumerGroupRefMut},
topic2::{TopicRef, TopicRefMut},
},
+ traits::MainOps,
},
};
use ahash::AHashMap;
-use iggy_common::{Identifier, IggyError};
+use iggy_common::{Identifier, IggyError, IggyTimestamp, PollingKind};
use slab::Slab;
-use std::{cell::RefCell, sync::Arc};
+use std::{
+ cell::RefCell,
+ sync::{Arc, atomic::Ordering},
+};
+use tracing::trace;
// Import streaming partitions helpers for the persist_messages method
use crate::streaming::partitions as streaming_partitions;
@@ -134,6 +140,227 @@ impl EntityComponentSystemMutCell for Streams {
}
}
+impl MainOps for Streams {
+ type Namespace = IggyFullNamespace;
+ type PollingArgs = PollingArgs;
+ type Consumer = PollingConsumer;
+ type In = IggyMessagesBatchMut;
+ type Out = (IggyPollMetadata, IggyMessagesBatchSet);
+ type Error = IggyError;
+
+ async fn append_messages(
+ &self,
+ shard_id: u16,
+ config: &SystemConfig,
+ ns: &Self::Namespace,
+ mut input: Self::In,
+ ) -> Result<(), Self::Error> {
+ let stream_id = ns.stream_id();
+ let topic_id = ns.topic_id();
+ let partition_id = ns.partition_id();
+
+ let current_offset = self.with_partition_by_id(
+ stream_id,
+ topic_id,
+ partition_id,
+ streaming_partitions::helpers::calculate_current_offset(),
+ );
+
+ let current_position = self.with_partition_by_id(stream_id, topic_id, partition_id, |(.., log)| {
+ log.journal().inner().size + log.active_segment().size
+ });
+ self.with_partition_by_id_async(
+ stream_id,
+ topic_id,
+ partition_id,
+ streaming_partitions::helpers::deduplicate_messages(current_offset, current_position, &mut input),
+ )
+ .await;
+
+ let (journal_messages_count, journal_size) = self.with_partition_by_id_mut(
+ stream_id,
+ topic_id,
+ partition_id,
+ streaming_partitions::helpers::append_to_journal(shard_id, current_offset, input),
+ )?;
+
+ let unsaved_messages_count_exceeded =
+ journal_messages_count >= config.partition.messages_required_to_save;
+ let unsaved_messages_size_exceeded = journal_size
+ >= config
+ .partition
+ .size_of_messages_required_to_save
+ .as_bytes_u64() as u32;
+
+ let is_full = self.with_partition_by_id(
+ stream_id,
+ topic_id,
+ partition_id,
+ streaming_partitions::helpers::is_segment_full(),
+ );
+
+ // Try committing the journal
+ if is_full || unsaved_messages_count_exceeded || unsaved_messages_size_exceeded {
+ self.persist_messages(
+ shard_id,
+ stream_id,
+ topic_id,
+ partition_id,
+ unsaved_messages_count_exceeded,
+ unsaved_messages_size_exceeded,
+ journal_messages_count,
+ journal_size,
+ config,
+ )
+ .await?;
+
+ if is_full {
+ self.handle_full_segment(shard_id, stream_id, topic_id, partition_id, config)
+ .await?;
+ }
+ }
+ Ok(())
+ }
+
+ async fn poll_messages(
+ &self,
+ ns: &Self::Namespace,
+ consumer: Self::Consumer,
+ args: Self::PollingArgs,
+ ) -> Result<Self::Out, Self::Error> {
+ let stream_id = ns.stream_id();
+ let topic_id = ns.topic_id();
+ let partition_id = ns.partition_id();
+ let current_offset = self.with_partition_by_id(
+ stream_id,
+ topic_id,
+ partition_id,
+ |(_, _, _, offset, ..)| offset.load(Ordering::Relaxed),
+ );
+ let metadata = IggyPollMetadata::new(partition_id as u32, current_offset);
+ let count = args.count;
+ let strategy = args.strategy;
+ let value = strategy.value;
+ let batches = match strategy.kind {
+ PollingKind::Offset => {
+ let offset = value;
+ // We have to remember to keep the invariant from the if that is on line 290.
+ // Alternatively a better design would be to get rid of that if and move the validations here.
+ if offset > current_offset {
+ return Ok((metadata, IggyMessagesBatchSet::default()));
+ }
+
+ let batches = self
+ .get_messages_by_offset(stream_id, topic_id, partition_id, offset, count)
+ .await?;
+ Ok(batches)
+ }
+ PollingKind::Timestamp => {
+ let timestamp = IggyTimestamp::from(value);
+ let timestamp_ts = timestamp.as_micros();
+ trace!(
+ "Getting {count} messages by timestamp: {} for partition: {}...",
+ timestamp_ts, partition_id
+ );
+
+ let batches = self
+ .get_messages_by_timestamp(
+ stream_id,
+ topic_id,
+ partition_id,
+ timestamp_ts,
+ count,
+ )
+ .await?;
+ Ok(batches)
+ }
+ PollingKind::First => {
+ let first_offset = self.with_partition_by_id(
+ stream_id,
+ topic_id,
+ partition_id,
+ |(_, _, _, _, _, _, log)| {
+ log.segments()
+ .first()
+ .map(|segment| segment.start_offset)
+ .unwrap_or(0)
+ },
+ );
+
+ let batches = self
+ .get_messages_by_offset(stream_id, topic_id, partition_id, first_offset, count)
+ .await?;
+ Ok(batches)
+ }
+ PollingKind::Last => {
+ let (start_offset, actual_count) = self.with_partition_by_id(
+ stream_id,
+ topic_id,
+ partition_id,
+ |(_, _, _, offset, _, _, _)| {
+ let current_offset = offset.load(Ordering::Relaxed);
+ let mut requested_count = count as u64;
+ if requested_count > current_offset + 1 {
+ requested_count = current_offset + 1
+ }
+ let start_offset = 1 + current_offset - requested_count;
+ (start_offset, requested_count as u32)
+ },
+ );
+
+ let batches = self
+ .get_messages_by_offset(
+ stream_id,
+ topic_id,
+ partition_id,
+ start_offset,
+ actual_count,
+ )
+ .await?;
+ Ok(batches)
+ }
+ PollingKind::Next => {
+ let (consumer_offset, consumer_id) = match consumer {
+ PollingConsumer::Consumer(consumer_id, _) => (
+ self.with_partition_by_id(
+ stream_id,
+ topic_id,
+ partition_id,
+ streaming_partitions::helpers::get_consumer_offset(consumer_id),
+ )
+ .map(|c_offset| c_offset.stored_offset),
+ consumer_id,
+ ),
+ PollingConsumer::ConsumerGroup(cg_id, _) => (
+ self.with_partition_by_id(
+ stream_id,
+ topic_id,
+ partition_id,
+ streaming_partitions::helpers::get_consumer_group_member_offset(cg_id),
+ )
+ .map(|cg_offset| cg_offset.stored_offset),
+ cg_id,
+ ),
+ };
+
+ let Some(consumer_offset) = consumer_offset else {
+ return Err(IggyError::ConsumerOffsetNotFound(consumer_id));
+ };
+ let offset = consumer_offset + 1;
+ trace!(
+ "Getting next messages for consumer id: {} for partition: {} from offset: {}...",
+ consumer_id, partition_id, offset
+ );
+ let batches = self
+ .get_messages_by_offset(stream_id, topic_id, partition_id, offset, count)
+ .await?;
+ Ok(batches)
+ }
+ }?;
+ Ok((metadata, batches))
+ }
+}
+
// A mental note:
// I think we can't expose as an access interface methods such as `get_topic_by_id` or `get_partition_by_id` etc..
// In a case of a `Stream` module replacement (with a new implementation), the new implementation might not have a notion of `Topic` or `Partition` at all.
@@ -430,12 +657,14 @@ impl Streams {
count: u32,
) -> Result<IggyMessagesBatchSet, IggyError> {
use crate::streaming::partitions::helpers;
- let range = self.with_partition_by_id(
+ let Ok(range) = self.with_partition_by_id(
stream_id,
topic_id,
partition_id,
helpers::get_segment_range_by_timestamp(timestamp),
- );
+ ) else {
+ return Ok(IggyMessagesBatchSet::default());
+ };
self.with_partition_by_id_async(
stream_id,
@@ -494,7 +723,7 @@ impl Streams {
)
});
- crate::shard_info!(
+ shard_info!(
shard_id,
"Closed segment for stream: {}, topic: {} with start offset: {}, end offset: {}, size: {} for partition with ID: {}.",
stream_id,
diff --git a/core/server/src/streaming/local_sizeable.rs b/core/server/src/streaming/local_sizeable.rs
deleted file mode 100644
index 813c1ffa..00000000
--- a/core/server/src/streaming/local_sizeable.rs
+++ /dev/null
@@ -1,31 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use iggy_common::IggyByteSize;
-
-/// Trait for types that return their size in bytes.
-/// repeated from the sdk because of the coherence rule, see messages.rs
-pub trait LocalSizeable {
- fn get_size_bytes(&self) -> IggyByteSize;
-}
-
-/// Trait for calculating the real memory size of a type, including all its fields
-/// and any additional overhead from containers like Arc, Vec, etc.
-pub trait RealSize {
- fn real_size(&self) -> IggyByteSize;
-}
diff --git a/core/server/src/streaming/mod.rs b/core/server/src/streaming/mod.rs
index 97215548..ee36cc4a 100644
--- a/core/server/src/streaming/mod.rs
+++ b/core/server/src/streaming/mod.rs
@@ -20,7 +20,6 @@ pub mod clients;
pub mod cluster;
pub mod deduplication;
pub mod diagnostics;
-pub mod local_sizeable;
pub mod partitions;
pub mod persistence;
pub mod personal_access_tokens;
@@ -31,5 +30,6 @@ pub mod stats;
pub mod storage;
pub mod streams;
pub mod topics;
+pub mod traits;
pub mod users;
pub mod utils;
diff --git a/core/server/src/streaming/partitions/helpers.rs b/core/server/src/streaming/partitions/helpers.rs
index bb37b568..8ce35655 100644
--- a/core/server/src/streaming/partitions/helpers.rs
+++ b/core/server/src/streaming/partitions/helpers.rs
@@ -295,7 +295,7 @@ pub async fn get_messages_by_offset(
} else {
0
};
-
+ let actual_first_offset_journal = journal.get(|batches| batches.first_offset()).unwrap();
let mut combined_batch_set = IggyMessagesBatchSet::empty();
// Load messages from disk if needed
@@ -316,10 +316,8 @@ pub async fn get_messages_by_offset(
if remaining_count > 0 {
let accumulator_start_offset = std::cmp::max(offset, journal_first_offset);
-
let accumulator_messages =
journal.get(|batches| batches.get_by_offset(accumulator_start_offset, remaining_count));
-
if !accumulator_messages.is_empty() {
combined_batch_set.add_batch_set(accumulator_messages);
}
@@ -462,10 +460,7 @@ pub fn get_segment_range_by_offset(
let start = log
.segments()
.iter()
- .enumerate()
- .filter(|(_, segment)| segment.start_offset <= offset)
- .map(|(index, _)| index)
- .next()
+ .rposition(|segment| segment.start_offset <= offset)
.expect("get_segment_range_by_offset: start segment not found");
let end = log.segments().len();
start..end
@@ -474,17 +469,22 @@ pub fn get_segment_range_by_offset(
pub fn get_segment_range_by_timestamp(
timestamp: u64,
-) -> impl FnOnce(ComponentsById<PartitionRef>) -> std::ops::Range<usize> {
- move |(.., log)| -> std::ops::Range<usize> {
+) -> impl FnOnce(ComponentsById<PartitionRef>) -> Result<std::ops::Range<usize>, IggyError> {
+ move |(.., log)| -> Result<std::ops::Range<usize>, IggyError> {
+ let segments = log.segments();
+ for seg in segments {
+ tracing::warn!("timestamp: {}, segment: {:?}", timestamp, seg);
+ }
let start = log
.segments()
.iter()
.enumerate()
- .find(|(_, segment)| segment.end_timestamp >= timestamp)
+ .filter(|(_, segment)| segment.end_timestamp >= timestamp)
.map(|(index, _)| index)
- .expect("get_segment_range_by_timestamp: start segment not found");
+ .next()
+ .ok_or(IggyError::TimestampOutOfRange(timestamp))?;
let end = log.segments().len();
- start..end
+ Ok(start..end)
}
}
@@ -644,6 +644,7 @@ pub fn calculate_current_offset() -> impl FnOnce(ComponentsById<PartitionRef>) -
pub fn deduplicate_messages(
current_offset: u64,
+ current_position: u32,
batch: &mut IggyMessagesBatchMut,
) -> impl AsyncFnOnce(ComponentsById<PartitionRef>) {
async move |(.., deduplicator, _, _, _, log)| {
@@ -652,7 +653,7 @@ pub fn deduplicate_messages(
.prepare_for_persistence(
segment.start_offset,
current_offset,
- segment.size,
+ current_position,
deduplicator.as_ref(),
)
.await;
@@ -704,6 +705,9 @@ pub fn commit_journal() -> impl FnOnce(ComponentsById<PartitionRefMut>) -> IggyM
let batches = log.journal_mut().commit();
log.ensure_indexes();
batches.append_indexes_to(log.active_indexes_mut().unwrap());
+ let indexes = log.active_indexes_mut().unwrap();
+ let first = indexes.get(0).unwrap();
+ let last = indexes.last().unwrap();
batches
}
}
@@ -777,8 +781,6 @@ pub fn persist_batch(
})?;
let indices = log.active_indexes().unwrap();
- let first_index = indices.get(0).unwrap();
- let last_index = indices.last().unwrap();
let unsaved_indexes_slice = log.active_indexes().unwrap().unsaved_slice();
let len = unsaved_indexes_slice.len();
storage
diff --git a/core/server/src/streaming/partitions/journal.rs b/core/server/src/streaming/partitions/journal.rs
index 9feee8de..54dfa225 100644
--- a/core/server/src/streaming/partitions/journal.rs
+++ b/core/server/src/streaming/partitions/journal.rs
@@ -79,7 +79,7 @@ impl Journal for MemoryMessageJournal {
}
fn commit(&mut self) -> Self::Container {
- self.inner.base_offset = self.inner.current_offset;
+ self.inner.base_offset = self.inner.current_offset + 1;
self.inner.first_timestamp = 0;
self.inner.end_timestamp = 0;
self.inner.size = 0;
diff --git a/core/server/src/streaming/partitions/partition2.rs b/core/server/src/streaming/partitions/partition2.rs
index 84cdfacb..a88bac02 100644
--- a/core/server/src/streaming/partitions/partition2.rs
+++ b/core/server/src/streaming/partitions/partition2.rs
@@ -1,15 +1,21 @@
use crate::{
+ configs::system::SystemConfig,
slab::{
partitions,
+ streams::Streams,
traits_ext::{EntityMarker, IntoComponents, IntoComponentsById},
},
streaming::{
+ self,
deduplication::message_deduplicator::MessageDeduplicator,
- partitions::{consumer_offset, journal::MemoryMessageJournal, log::SegmentedLog},
- stats::stats::PartitionStats,
+ partitions::{
+ consumer_offset, helpers::create_message_deduplicator, journal::MemoryMessageJournal,
+ log::SegmentedLog,
+ },
+ stats::stats::{PartitionStats, TopicStats},
},
};
-use iggy_common::IggyTimestamp;
+use iggy_common::{Identifier, IggyTimestamp};
use slab::Slab;
use std::sync::{Arc, atomic::AtomicU64};
@@ -365,3 +371,45 @@ impl<'a> IntoComponentsById for PartitionRefMut<'a> {
)
}
}
+
+pub fn create_and_insert_partitions_mem(
+ streams: &Streams,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ parent_stats: Arc<TopicStats>,
+ partitions_count: u32,
+ config: &SystemConfig,
+) -> Vec<Partition> {
+ let range = 0..partitions_count as usize;
+ let created_at = IggyTimestamp::now();
+ range
+ .map(|_| {
+ // Areczkuuuu.
+ let stats = Arc::new(PartitionStats::new(parent_stats.clone()));
+ let should_increment_offset = false;
+ let deduplicator = create_message_deduplicator(config);
+ let offset = Arc::new(AtomicU64::new(0));
+ let consumer_offset = Arc::new(ConsumerOffsets::with_capacity(2137));
+ let consumer_group_offset = Arc::new(ConsumerGroupOffsets::with_capacity(2137));
+ let log = Default::default();
+
+ let mut partition = Partition::new(
+ created_at,
+ should_increment_offset,
+ stats,
+ deduplicator,
+ offset,
+ consumer_offset,
+ consumer_group_offset,
+ log,
+ );
+ let id = streams.with_partitions_mut(
+ stream_id,
+ topic_id,
+ streaming::partitions::helpers::insert_partition(partition.clone()),
+ );
+ partition.update_id(id);
+ partition
+ })
+ .collect()
+}
diff --git a/core/server/src/streaming/segments/types/messages_batch_set.rs b/core/server/src/streaming/segments/types/messages_batch_set.rs
index 285c5174..42c8747c 100644
--- a/core/server/src/streaming/segments/types/messages_batch_set.rs
+++ b/core/server/src/streaming/segments/types/messages_batch_set.rs
@@ -212,14 +212,6 @@ impl IggyMessagesBatchSet {
if self.is_empty() || count == 0 {
return Self::empty();
}
- tracing::trace!(
- "Getting {} messages from batch set, start offset {}, end offset calculated {}, end offset real {}, messages count {}...",
- count,
- start_offset,
- start_offset + count as u64 - 1,
- self.last_offset().unwrap_or(0),
- self.count()
- );
let mut result = Self::with_capacity(self.containers_count());
let mut remaining_count = count;
diff --git a/core/server/src/streaming/streams/messages.rs b/core/server/src/streaming/streams/messages.rs
deleted file mode 100644
index 679c9938..00000000
--- a/core/server/src/streaming/streams/messages.rs
+++ /dev/null
@@ -1,111 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use error_set::ErrContext;
-use iggy_common::{Consumer, Identifier, IggyError, Partitioning};
-use tracing::trace;
-
-use super::COMPONENT;
-use crate::{
- binary::handlers::messages::poll_messages_handler::IggyPollMetadata,
- shard::transmission::message::{ShardMessage, ShardRequest, ShardRequestPayload},
- streaming::{
- polling_consumer::PollingConsumer,
- segments::{IggyMessagesBatchMut, IggyMessagesBatchSet},
- },
-};
-use std::sync::atomic::Ordering;
-
-/*
-impl Stream {
- pub fn get_messages_count(&self) -> u64 {
- self.messages_count.load(Ordering::SeqCst)
- }
-
- // TODO: Create a types module, where we will put types of those closures.
- pub async fn append_messages(
- &self,
- topic_id: &Identifier,
- partitioning: &Partitioning,
- append_messages: impl AsyncFnOnce(&Topic, u32) -> Result<(), IggyError>,
- ) -> Result<(), IggyError> {
- // It's quite empty there, but in the future if we would like to introduce consumer group subscription
- // based on topic name wildcards, we would have to lift those to stream level,
- // thus it would be perfect place to perform those calculations here.
- let topic = self.get_topic(topic_id).with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to get topic with ID: {} for stream ID: {}",
- topic_id, self.stream_id
- )
- })?;
-
- let partition_id = topic.calculate_partition_id(partitioning)?;
- append_messages(topic, partition_id).await
- }
-
- pub async fn poll_messages(
- &self,
- topic_id: &Identifier,
- client_id: u32,
- consumer: Consumer,
- maybe_partition_id: Option<u32>,
- auto_commit: bool,
- poll_messages: impl AsyncFnOnce(
- &Topic,
- PollingConsumer,
- u32,
- )
- -> Result<(IggyPollMetadata, IggyMessagesBatchSet), IggyError>,
- ) -> Result<(IggyPollMetadata, IggyMessagesBatchSet), IggyError> {
- // Same as in `append_messages` method.
- let topic = self.get_topic(topic_id).with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to get topic with ID: {} for stream ID: {}",
- topic_id, self.stream_id
- )
- })?;
- if !topic.has_partitions() {
- return Err(IggyError::NoPartitions(topic.topic_id, self.stream_id));
- }
-
- let Some((consumer, partition_id)) = topic
- .resolve_consumer_with_partition_id(&consumer, client_id, maybe_partition_id, true)
- .with_error_context(|error| format!("{COMPONENT} (error: {error}) - failed to resolve consumer with partition id, consumer: {consumer}, client ID: {}, partition ID: {:?}", client_id, maybe_partition_id))? else {
- todo!("Send early response");
- };
-
- let (metadata, batch) = poll_messages(topic, consumer, partition_id).await?;
-
- if auto_commit && !batch.is_empty() {
- let offset = batch
- .last_offset()
- .expect("Batch set should have at least one batch");
- trace!(
- "Last offset: {} will be automatically stored for {}, stream: {}, topic: {}, partition: {}",
- offset, consumer, self.stream_id, topic_id, partition_id
- );
- topic
- .store_consumer_offset_internal(consumer, offset, partition_id)
- .await
- .with_error_context(|error| format!("{COMPONENT} (error: {error}) - failed to store consumer offset internal, polling consumer: {consumer}, offset: {offset}, partition ID: {partition_id}"))?;
- }
- Ok((metadata, batch))
- }
-}
-
-*/
diff --git a/core/server/src/streaming/streams/mod.rs b/core/server/src/streaming/streams/mod.rs
index 73031f0b..044a2b63 100644
--- a/core/server/src/streaming/streams/mod.rs
+++ b/core/server/src/streaming/streams/mod.rs
@@ -17,7 +17,6 @@
*/
pub mod helpers;
-pub mod messages;
pub mod storage;
pub mod storage2;
pub mod stream2;
diff --git a/core/server/src/streaming/streams/stream2.rs b/core/server/src/streaming/streams/stream2.rs
index f0c88455..098ed8fc 100644
--- a/core/server/src/streaming/streams/stream2.rs
+++ b/core/server/src/streaming/streams/stream2.rs
@@ -7,9 +7,10 @@ use std::{
use crate::{
slab::{
- Keyed, streams,
+ Keyed,
+ streams::{self, Streams},
topics::Topics,
- traits_ext::{EntityMarker, IntoComponents, IntoComponentsById},
+ traits_ext::{EntityMarker, InsertCell, IntoComponents, IntoComponentsById},
},
streaming::stats::stats::StreamStats,
};
@@ -187,3 +188,13 @@ impl<'a> IntoComponentsById for StreamRefMut<'a> {
(root, stats)
}
}
+
+// TODO: Move this to a dedicated module.
+pub fn create_and_insert_stream_mem(streams: &Streams, name: String) -> Stream {
+ let now = IggyTimestamp::now();
+ let stats = Arc::new(Default::default());
+ let mut stream = Stream::new(name, stats, now);
+ let id = streams.insert(stream.clone());
+ stream.update_id(id);
+ stream
+}
diff --git a/core/server/src/streaming/topics/messages.rs b/core/server/src/streaming/topics/messages.rs
deleted file mode 100644
index 5779db1a..00000000
--- a/core/server/src/streaming/topics/messages.rs
+++ /dev/null
@@ -1,362 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::binary::handlers::messages::poll_messages_handler::IggyPollMetadata;
-use crate::streaming::polling_consumer::PollingConsumer;
-use crate::streaming::segments::{IggyMessagesBatchMut, IggyMessagesBatchSet};
-use crate::streaming::topics::COMPONENT;
-use crate::streaming::utils::hash;
-use ahash::AHashMap;
-use error_set::ErrContext;
-use iggy_common::locking::IggyRwLockFn;
-use iggy_common::{IggyError, IggyExpiry, Partitioning, PartitioningKind, PollingKind};
-use iggy_common::{IggyTimestamp, PollingStrategy};
-use std::sync::atomic::Ordering;
-use tracing::trace;
-
-/*
-
-impl Topic {
- pub fn get_messages_count(&self) -> u64 {
- self.messages_count.load(Ordering::SeqCst)
- }
-
- pub async fn get_messages(
- &self,
- consumer: PollingConsumer,
- partition_id: u32,
- strategy: PollingStrategy,
- count: u32,
- ) -> Result<(IggyPollMetadata, IggyMessagesBatchSet), IggyError> {
- if !self.has_partitions() {
- return Err(IggyError::NoPartitions(self.topic_id, self.stream_id));
- }
-
- let partition = self.partitions.get(&partition_id);
- if partition.is_none() {
- return Err(IggyError::PartitionNotFound(
- partition_id,
- self.topic_id,
- self.stream_id,
- ));
- }
-
- let partition = partition.unwrap();
- let partition = partition.read().await;
- let value = strategy.value;
- let messages = match strategy.kind {
- PollingKind::Offset => partition.get_messages_by_offset(value, count).await,
- PollingKind::Timestamp => {
- partition
- .get_messages_by_timestamp(value.into(), count)
- .await
- .with_error_context(|error| format!("{COMPONENT} (error: {error}) - failed to get messages by timestamp: {value}, count: {count}"))
- }
- PollingKind::First => partition.get_first_messages(count).await,
- PollingKind::Last => partition.get_last_messages(count).await,
- PollingKind::Next => partition.get_next_messages(consumer, count).await,
- }?;
-
- let metadata = IggyPollMetadata::new(partition_id, partition.current_offset);
-
- Ok((metadata, messages))
- }
-
- pub async fn append_messages(
- &self,
- partition_id: u32,
- batch: IggyMessagesBatchMut,
- ) -> Result<(), IggyError> {
- if !self.has_partitions() {
- return Err(IggyError::NoPartitions(self.topic_id, self.stream_id));
- }
-
- // Don't return an error if the topic is full and delete_oldest_segments is true.
- // Oldest segment will be removed eventually by MaintainMessages background job.
- if self.is_full() && self.config.topic.delete_oldest_segments {
- return Err(IggyError::TopicFull(self.topic_id, self.stream_id));
- }
-
- if batch.is_empty() {
- return Ok(());
- }
-
- self.append_messages_to_partition(batch, partition_id).await
- }
-
- pub async fn flush_unsaved_buffer(
- &self,
- partition_id: u32,
- fsync: bool,
- ) -> Result<(), IggyError> {
- let partition = self.partitions.get(&partition_id);
- partition
- .ok_or(IggyError::PartitionNotFound(
- partition_id,
- self.stream_id,
- self.stream_id,
- ))?
- .write()
- .await
- .flush_unsaved_buffer(fsync)
- .await
- }
-
- async fn append_messages_to_partition(
- &self,
- messages: IggyMessagesBatchMut,
- partition_id: u32,
- ) -> Result<(), IggyError> {
- let partition = self.partitions.get(&partition_id);
- partition
- .ok_or(IggyError::PartitionNotFound(
- partition_id,
- self.topic_id,
- self.stream_id,
- ))?
- .write()
- .await
- .append_messages(messages)
- .await
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to append messages")
- })?;
-
- Ok(())
- }
-
- pub fn calculate_partition_id(&self, partitioning: &Partitioning) -> Result<u32, IggyError> {
- match partitioning.kind {
- PartitioningKind::Balanced => Ok(self.get_next_partition_id()),
- PartitioningKind::PartitionId => Ok(u32::from_le_bytes(
- partitioning.value[..partitioning.length as usize]
- .try_into()
- .map_err(|_| IggyError::InvalidNumberEncoding)?,
- )),
- PartitioningKind::MessagesKey => {
- Ok(self.calculate_partition_id_by_messages_key_hash(&partitioning.value))
- }
- }
- }
-
- fn get_next_partition_id(&self) -> u32 {
- let mut partition_id = self.current_partition_id.fetch_add(1, Ordering::SeqCst);
- let partitions_count = self.partitions.len() as u32;
- if partition_id > partitions_count {
- partition_id = 1;
- self.current_partition_id
- .swap(partition_id + 1, Ordering::SeqCst);
- }
- trace!("Next partition ID: {}", partition_id);
- partition_id
- }
-
- fn calculate_partition_id_by_messages_key_hash(&self, messages_key: &[u8]) -> u32 {
- let messages_key_hash = hash::calculate_32(messages_key);
- let partitions_count = self.get_partitions_count();
- let mut partition_id = messages_key_hash % partitions_count;
- if partition_id == 0 {
- partition_id = partitions_count;
- }
- trace!(
- "Calculated partition ID: {} for messages key: {:?}, hash: {}",
- partition_id, messages_key, messages_key_hash
- );
- partition_id
- }
-
- pub async fn get_expired_segments_start_offsets_per_partition(
- &self,
- now: IggyTimestamp,
- ) -> AHashMap<u32, Vec<u64>> {
- let mut expired_segments = AHashMap::new();
- if let IggyExpiry::ExpireDuration(_) = self.message_expiry {
- for (_, partition) in self.partitions.iter() {
- let partition = partition.read().await;
- let segments = partition.get_expired_segments_start_offsets(now).await;
- if !segments.is_empty() {
- expired_segments.insert(partition.partition_id, segments);
- }
- }
- }
- expired_segments
- }
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use crate::configs::system::SystemConfig;
- use crate::streaming::persistence::persister::FileWithSyncPersister;
- use crate::streaming::persistence::persister::PersisterKind;
- use crate::streaming::storage::SystemStorage;
- use crate::streaming::utils::MemoryPool;
- use bytes::Bytes;
- use iggy_common::CompressionAlgorithm;
- use iggy_common::{IggyMessage, MaxTopicSize};
- use std::rc::Rc;
- use std::sync::Arc;
- use std::sync::atomic::AtomicU32;
- use std::sync::atomic::AtomicU64;
-
- #[tokio::test]
- async fn given_partition_id_key_messages_should_be_appended_only_to_the_chosen_partition() {
- let partition_id = 1;
- let partitioning = Partitioning::partition_id(partition_id);
- let partitions_count = 3;
- let messages_count: u32 = 1000;
- let topic = init_topic(partitions_count).await;
-
- for entity_id in 1..=messages_count {
- let message = IggyMessage::builder()
- .id(entity_id as u128)
- .payload(Bytes::from(entity_id.to_string()))
- .build()
- .expect("Failed to create message with valid payload and headers");
- let messages = IggyMessagesBatchMut::from_messages(&[message], 1);
- let partition = topic.calculate_partition_id(&partitioning).unwrap();
- topic.append_messages(partition, messages).await.unwrap();
- }
-
- let partitions = topic.get_partitions();
- assert_eq!(partitions.len(), partitions_count as usize);
- for partition in partitions {
- let partition = partition.read().await;
- let partition_msgs_count = partition.get_messages_count();
- if partition.partition_id == partition_id {
- assert_eq!(partition_msgs_count, messages_count as u64);
- } else {
- assert_eq!(partition_msgs_count, 0);
- }
- }
- }
-
- #[tokio::test]
- async fn given_messages_key_key_messages_should_be_appended_to_the_calculated_partitions() {
- let partitions_count = 3;
- let messages_count = 1000;
- let topic = init_topic(partitions_count).await;
-
- for entity_id in 1..=messages_count {
- let partitioning = Partitioning::messages_key_u32(entity_id);
- let message = IggyMessage::builder()
- .id(entity_id as u128)
- .payload(Bytes::from("test message"))
- .build()
- .expect("Failed to create message with valid payload and headers");
- let messages = IggyMessagesBatchMut::from_messages(&[message], 1);
- let partition_id = topic
- .calculate_partition_id(&partitioning)
- .expect("Failed to calculate partition ID");
- topic.append_messages(partition_id, messages).await.unwrap();
- }
-
- let mut read_messages_count = 0;
- let partitions = topic.get_partitions();
- assert_eq!(partitions.len(), partitions_count as usize);
- for partition in partitions {
- let partition = partition.read().await;
- let partition_msgs_cnt = partition.get_messages_count();
- read_messages_count += partition_msgs_cnt;
- assert!(partition_msgs_cnt < messages_count as u64);
- }
-
- assert_eq!(read_messages_count, messages_count as u64);
- }
-
- #[tokio::test]
- async fn given_multiple_partitions_calculate_next_partition_id_should_return_next_partition_id_using_round_robin()
- {
- let partitions_count = 3;
- let messages_count = 1000;
- let topic = init_topic(partitions_count).await;
-
- let mut expected_partition_id = 0;
- for _ in 1..=messages_count {
- let partition_id = topic.get_next_partition_id();
- expected_partition_id += 1;
- if expected_partition_id > partitions_count {
- expected_partition_id = 1;
- }
-
- assert_eq!(partition_id, expected_partition_id);
- }
- }
-
- #[tokio::test]
- async fn given_multiple_partitions_calculate_partition_id_by_hash_should_return_next_partition_id()
- {
- let partitions_count = 3;
- let messages_count = 1000;
- let topic = init_topic(partitions_count).await;
-
- for entity_id in 1..=messages_count {
- let key = Partitioning::messages_key_u32(entity_id);
- let partition_id = topic.calculate_partition_id_by_messages_key_hash(&key.value);
- let entity_id_hash = hash::calculate_32(&key.value);
- let mut expected_partition_id = entity_id_hash % partitions_count;
- if expected_partition_id == 0 {
- expected_partition_id = partitions_count;
- }
-
- assert_eq!(partition_id, expected_partition_id);
- }
- }
-
- async fn init_topic(partitions_count: u32) -> Topic {
- let tempdir = tempfile::TempDir::new().unwrap();
- let config = Arc::new(SystemConfig {
- path: tempdir.path().to_str().unwrap().to_string(),
- ..Default::default()
- });
- let storage = Rc::new(SystemStorage::new(
- config.clone(),
- Arc::new(PersisterKind::FileWithSync(FileWithSyncPersister {})),
- ));
- MemoryPool::init_pool(config.clone());
- let stream_id = 1;
- let id = 2;
- let name = "test";
- let compression_algorithm = CompressionAlgorithm::None;
- let size_of_parent_stream = Arc::new(AtomicU64::new(0));
- let messages_count_of_parent_stream = Arc::new(AtomicU64::new(0));
- let segments_count_of_parent_stream = Arc::new(AtomicU32::new(0));
-
- let created_info = Topic::create(
- stream_id,
- id,
- name,
- partitions_count,
- config,
- storage,
- size_of_parent_stream,
- messages_count_of_parent_stream,
- segments_count_of_parent_stream,
- IggyExpiry::NeverExpire,
- compression_algorithm,
- MaxTopicSize::ServerDefault,
- 1,
- )
- .unwrap();
- let topic = created_info.topic;
- topic.persist().await.unwrap();
- topic
- }
-}
-
-*/
diff --git a/core/server/src/streaming/topics/mod.rs b/core/server/src/streaming/topics/mod.rs
index e3700db0..88028a71 100644
--- a/core/server/src/streaming/topics/mod.rs
+++ b/core/server/src/streaming/topics/mod.rs
@@ -18,7 +18,6 @@
pub mod consumer_group2;
pub mod helpers;
-pub mod messages;
pub mod storage2;
pub mod topic2;
diff --git a/core/server/src/streaming/topics/storage2.rs b/core/server/src/streaming/topics/storage2.rs
index d6b331cb..35e854a7 100644
--- a/core/server/src/streaming/topics/storage2.rs
+++ b/core/server/src/streaming/topics/storage2.rs
@@ -39,23 +39,6 @@ pub async fn create_topic_file_hierarchy(
topic_id as u32,
));
}
-
- /*
- 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/topic2.rs b/core/server/src/streaming/topics/topic2.rs
index a5ddf813..e2ae830c 100644
--- a/core/server/src/streaming/topics/topic2.rs
+++ b/core/server/src/streaming/topics/topic2.rs
@@ -1,11 +1,14 @@
use crate::configs::system::SystemConfig;
use crate::shard_trace;
+use crate::slab::streams::Streams;
use crate::slab::topics;
-use crate::slab::traits_ext::{EntityMarker, IntoComponents, IntoComponentsById};
+use crate::slab::traits_ext::{EntityMarker, InsertCell, IntoComponents, IntoComponentsById};
use crate::slab::{Keyed, consumer_groups::ConsumerGroups, partitions::Partitions};
use crate::streaming::partitions::log::SegmentedLog;
-use crate::streaming::stats::stats::TopicStats;
-use iggy_common::{CompressionAlgorithm, IggyError, IggyExpiry, IggyTimestamp, MaxTopicSize};
+use crate::streaming::stats::stats::{StreamStats, TopicStats};
+use iggy_common::{
+ CompressionAlgorithm, Identifier, IggyError, IggyExpiry, IggyTimestamp, MaxTopicSize,
+};
use slab::Slab;
use std::cell::{Ref, RefMut};
use std::sync::Arc;
@@ -319,28 +322,30 @@ impl TopicRoot {
}
}
-pub fn resolve_max_topic_size(
+// TODO: Move to separate module.
+pub fn create_and_insert_topics_mem(
+ streams: &Streams,
+ stream_id: &Identifier,
+ name: String,
+ replication_factor: u8,
+ message_expiry: IggyExpiry,
+ compression: CompressionAlgorithm,
max_topic_size: MaxTopicSize,
- config: &SystemConfig,
-) -> Result<MaxTopicSize, IggyError> {
- match max_topic_size {
- MaxTopicSize::ServerDefault => Ok(config.topic.max_size),
- _ => {
- if max_topic_size.as_bytes_u64() < config.segment.size.as_bytes_u64() {
- Err(IggyError::InvalidTopicSize(
- max_topic_size,
- config.segment.size,
- ))
- } else {
- Ok(max_topic_size)
- }
- }
- }
-}
+ parent_stats: Arc<StreamStats>,
+) -> Topic {
+ let stats = Arc::new(TopicStats::new(parent_stats));
+ let now = IggyTimestamp::now();
+ let mut topic = Topic::new(
+ name,
+ stats,
+ now,
+ replication_factor,
+ message_expiry,
+ compression,
+ max_topic_size,
+ );
-pub fn resolve_message_expiry(message_expiry: IggyExpiry, config: &SystemConfig) -> IggyExpiry {
- match message_expiry {
- IggyExpiry::ServerDefault => config.segment.message_expiry,
- _ => message_expiry,
- }
+ let id = streams.with_topics(stream_id, |topics| topics.insert(topic.clone()));
+ topic.update_id(id);
+ topic
}
diff --git a/core/server/src/streaming/traits.rs b/core/server/src/streaming/traits.rs
new file mode 100644
index 00000000..478c7f50
--- /dev/null
+++ b/core/server/src/streaming/traits.rs
@@ -0,0 +1,25 @@
+use crate::configs::system::SystemConfig;
+
+// TODO: Major revision of this trait.
+pub trait MainOps {
+ type Namespace;
+ type PollingArgs;
+ type Consumer;
+ type In;
+ type Out;
+ type Error;
+
+ fn append_messages(
+ &self,
+ shard_id: u16,
+ config: &SystemConfig,
+ ns: &Self::Namespace,
+ input: Self::In,
+ ) -> impl Future<Output = Result<(), Self::Error>>;
+ fn poll_messages(
+ &self,
+ ns: &Self::Namespace,
+ consumer: Self::Consumer,
+ args: Self::PollingArgs,
+ ) -> impl Future<Output = Result<Self::Out, Self::Error>>;
+}