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 19:20:23 UTC
(iggy) branch fix_integration_tests_for_messages created (now 2add00a0)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch fix_integration_tests_for_messages
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 2add00a0 feat(io_uring): fix integration tests for messages
This branch includes the following new commits:
new 2add00a0 feat(io_uring): fix integration tests for messages
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: feat(io_uring): fix integration tests for messages
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch fix_integration_tests_for_messages
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 2add00a04577e61199c8960364385e06660aee22
Author: numminex <g....@gmail.com>
AuthorDate: Mon Sep 15 21:20:11 2025 +0200
feat(io_uring): fix integration tests for messages
---
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 | 251 ++++++++-
core/server/src/streaming/local_sizeable.rs | 31 --
core/server/src/streaming/mod.rs | 2 +-
core/server/src/streaming/partitions/helpers.rs | 17 +-
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 +
40 files changed, 725 insertions(+), 3203 deletions(-)
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..77855634 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.
@@ -494,7 +721,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..7ef753e7 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
@@ -644,6 +639,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 +648,7 @@ pub fn deduplicate_messages(
.prepare_for_persistence(
segment.start_offset,
current_offset,
- segment.size,
+ current_position,
deduplicator.as_ref(),
)
.await;
@@ -704,6 +700,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 +776,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>>;
+}