You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2025/04/05 23:26:51 UTC
(iggy) 02/03: improve performance on send
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch zero-copy-no-batching-rebase
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 66a7a0172a0d04727befe02b39b7023daf5d6c47
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Sun Apr 6 01:26:22 2025 +0200
improve performance on send
---
sdk/src/models/messaging/messages_batch.rs | 2 +-
server/src/streaming/partitions/segments.rs | 1 -
.../src/streaming/segments/indexes/index_reader.rs | 39 +++--
.../src/streaming/segments/indexes/index_writer.rs | 32 ++--
.../streaming/segments/messages/messages_reader.rs | 163 +++++----------------
.../streaming/segments/messages/messages_writer.rs | 31 ++--
.../src/streaming/segments/messages_accumulator.rs | 9 +-
server/src/streaming/segments/reading_messages.rs | 4 +-
server/src/streaming/segments/segment.rs | 20 ++-
.../streaming/segments/types/messages_batch_mut.rs | 13 +-
.../streaming/segments/types/messages_batch_set.rs | 2 -
server/src/streaming/segments/writing_messages.rs | 15 +-
12 files changed, 140 insertions(+), 191 deletions(-)
diff --git a/sdk/src/models/messaging/messages_batch.rs b/sdk/src/models/messaging/messages_batch.rs
index b717d6dc..0dd0b3ba 100644
--- a/sdk/src/models/messaging/messages_batch.rs
+++ b/sdk/src/models/messaging/messages_batch.rs
@@ -117,7 +117,7 @@ impl IggyMessagesBatch {
return None;
}
- // TODO(hubcio): this can be optimized via binary search
+ // TODO(hubcio): this can be optimized via lookup
let first_message_index = self
.iter()
.position(|msg| msg.header().offset() >= start_offset);
diff --git a/server/src/streaming/partitions/segments.rs b/server/src/streaming/partitions/segments.rs
index 687bc217..296106d7 100644
--- a/server/src/streaming/partitions/segments.rs
+++ b/server/src/streaming/partitions/segments.rs
@@ -69,7 +69,6 @@ impl Partition {
self.messages_count_of_parent_topic.clone(),
self.messages_count.clone(),
);
-
new_segment.persist().await.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to persist new segment: {new_segment}",)
})?;
diff --git a/server/src/streaming/segments/indexes/index_reader.rs b/server/src/streaming/segments/indexes/index_reader.rs
index 98b4fd52..31ad4420 100644
--- a/server/src/streaming/segments/indexes/index_reader.rs
+++ b/server/src/streaming/segments/indexes/index_reader.rs
@@ -4,7 +4,7 @@ use error_set::ErrContext;
use iggy::models::messaging::{IggyIndex, IggyIndexes, INDEX_SIZE};
use iggy::{error::IggyError, models::messaging::IggyIndexView};
use std::{
- fs::{File, OpenOptions},
+ fs::File as StdFile,
io::ErrorKind,
os::unix::fs::FileExt,
sync::{
@@ -12,6 +12,7 @@ use std::{
Arc,
},
};
+use tokio::fs::OpenOptions;
use tokio::task::spawn_blocking;
use tracing::{error, trace};
@@ -19,7 +20,7 @@ use tracing::{error, trace};
#[derive(Debug)]
pub struct IndexReader {
file_path: String,
- file: Arc<File>,
+ file: Arc<StdFile>,
index_size_bytes: Arc<AtomicU64>,
}
@@ -29,23 +30,35 @@ impl IndexReader {
let file = OpenOptions::new()
.read(true)
.open(file_path)
+ .await
.with_error_context(|error| format!("Failed to open index file: {file_path}. {error}"))
.map_err(|_| IggyError::CannotReadFile)?;
- let actual_index_size = file
- .metadata()
- .with_error_context(|error| {
- format!("Failed to get metadata of index file: {file_path}. {error}")
- })
- .map_err(|_| IggyError::CannotReadFileMetadata)?
- .len();
-
- index_size_bytes.store(actual_index_size, Ordering::Release);
+ // posix_fadvise() doesn't exist on MacOS
+ #[cfg(not(target_os = "macos"))]
+ {
+ use std::os::unix::io::AsRawFd;
+ let fd = file.as_raw_fd();
+ let _ = nix::fcntl::posix_fadvise(
+ fd,
+ 0,
+ 0, // 0 means the entire file
+ nix::fcntl::PosixFadviseAdvice::POSIX_FADV_SEQUENTIAL,
+ )
+ .with_info_context(|error| {
+ format!(
+ "Failed to set sequential access pattern on index file: {file_path}. {error}"
+ )
+ });
+ }
- trace!("Opened index file for reading: {file_path}, size: {actual_index_size}",);
+ trace!(
+ "Opened index file for reading: {file_path}, size: {}",
+ index_size_bytes.load(Ordering::Acquire)
+ );
Ok(Self {
file_path: file_path.to_string(),
- file: Arc::new(file),
+ file: Arc::new(file.into_std().await),
index_size_bytes,
})
}
diff --git a/server/src/streaming/segments/indexes/index_writer.rs b/server/src/streaming/segments/indexes/index_writer.rs
index 527ab050..dacfb9e9 100644
--- a/server/src/streaming/segments/indexes/index_writer.rs
+++ b/server/src/streaming/segments/indexes/index_writer.rs
@@ -26,6 +26,7 @@ impl IndexWriter {
file_path: &str,
index_size_bytes: Arc<AtomicU64>,
fsync: bool,
+ file_exists: bool,
) -> Result<Self, IggyError> {
let file = OpenOptions::new()
.write(true)
@@ -36,22 +37,27 @@ impl IndexWriter {
.with_error_context(|error| format!("Failed to open index file: {file_path}. {error}"))
.map_err(|_| IggyError::CannotReadFile)?;
- let _ = file.sync_all().await.with_error_context(|error| {
- format!("Failed to fsync index file after creation: {file_path}. {error}",)
- });
+ if file_exists {
+ let _ = file.sync_all().await.with_error_context(|error| {
+ format!("Failed to fsync index file after creation: {file_path}. {error}",)
+ });
- let actual_index_size = file
- .metadata()
- .await
- .with_error_context(|error| {
- format!("Failed to get metadata of index file: {file_path}. {error}")
- })
- .map_err(|_| IggyError::CannotReadFileMetadata)?
- .len();
+ let actual_index_size = file
+ .metadata()
+ .await
+ .with_error_context(|error| {
+ format!("Failed to get metadata of index file: {file_path}. {error}")
+ })
+ .map_err(|_| IggyError::CannotReadFileMetadata)?
+ .len();
- index_size_bytes.store(actual_index_size, Ordering::Release);
+ index_size_bytes.store(actual_index_size, Ordering::Release);
+ }
- trace!("Opened index file for writing: {file_path}, size: {actual_index_size}");
+ trace!(
+ "Opened index file for writing: {file_path}, size: {}",
+ index_size_bytes.load(Ordering::Acquire)
+ );
Ok(Self {
file_path: file_path.to_string(),
diff --git a/server/src/streaming/segments/messages/messages_reader.rs b/server/src/streaming/segments/messages/messages_reader.rs
index 650a3980..fa7345b7 100644
--- a/server/src/streaming/segments/messages/messages_reader.rs
+++ b/server/src/streaming/segments/messages/messages_reader.rs
@@ -3,13 +3,8 @@ use error_set::ErrContext;
use iggy::{
error::IggyError,
models::messaging::{IggyIndexes, IggyMessagesBatch},
- prelude::IggyTimestamp,
-};
-use std::{
- fs::{File, OpenOptions},
- os::{fd::AsRawFd, unix::prelude::FileExt},
- time::Duration,
};
+use std::{fs::File as StdFile, os::unix::prelude::FileExt};
use std::{
io::ErrorKind,
sync::{
@@ -17,38 +12,32 @@ use std::{
Arc,
},
};
-use tokio::{
- task::{spawn_blocking, JoinHandle},
- time::interval,
-};
-use tracing::{debug, error, trace};
+use tokio::fs::OpenOptions;
+use tokio::task::spawn_blocking;
+use tracing::{error, trace};
-/// A dedicated struct for reading from the log file.
+/// A dedicated struct for reading from the messages file.
#[derive(Debug)]
pub struct MessagesReader {
file_path: String,
- file: Arc<File>,
- log_size_bytes: Arc<AtomicU64>,
- last_access: Arc<AtomicU64>,
- cache_advisor_task: Option<JoinHandle<()>>,
+ file: Arc<StdFile>,
+ messages_size_bytes: Arc<AtomicU64>,
}
impl MessagesReader {
- /// Opens the log file in read mode.
- pub async fn new(file_path: &str, log_size_bytes: Arc<AtomicU64>) -> Result<Self, IggyError> {
+ /// Opens the messages file in read mode.
+ pub async fn new(
+ file_path: &str,
+ messages_size_bytes: Arc<AtomicU64>,
+ ) -> Result<Self, IggyError> {
let file = OpenOptions::new()
.read(true)
.open(file_path)
- .with_error_context(|error| format!("Failed to open log file: {file_path}. {error}"))
- .map_err(|_| IggyError::CannotReadFile)?;
-
- let actual_log_size = file
- .metadata()
+ .await
.with_error_context(|error| {
- format!("Failed to get metadata of log file: {file_path}. {error}")
+ format!("Failed to open messages file: {file_path}, error: {error}")
})
- .map_err(|_| IggyError::CannotReadFileMetadata)?
- .len();
+ .map_err(|_| IggyError::CannotReadFile)?;
// posix_fadvise() doesn't exist on MacOS
#[cfg(not(target_os = "macos"))]
@@ -56,33 +45,31 @@ impl MessagesReader {
use std::os::unix::io::AsRawFd;
let fd = file.as_raw_fd();
let _ = nix::fcntl::posix_fadvise(
- fd,
- 0,
- actual_log_size as i64,
- nix::fcntl::PosixFadviseAdvice::POSIX_FADV_SEQUENTIAL,
- )
- .with_info_context(|error| {
- format!("Failed to set sequential access pattern on log file: {file_path}. {error}")
- });
+ fd,
+ 0,
+ 0, // 0 means the entire file
+ nix::fcntl::PosixFadviseAdvice::POSIX_FADV_SEQUENTIAL,
+ )
+ .with_info_context(|error| {
+ format!(
+ "Failed to set sequential access pattern on messages file: {file_path}. {error}"
+ )
+ });
}
- log_size_bytes.store(actual_log_size, Ordering::Release);
-
- trace!("Opened messages file for reading: {file_path}, size: {actual_log_size}");
+ trace!(
+ "Opened messages file for reading: {file_path}, size: {}",
+ messages_size_bytes.load(Ordering::Acquire)
+ );
- let mut reader = Self {
+ Ok(Self {
file_path: file_path.to_string(),
- file: Arc::new(file),
- log_size_bytes,
- last_access: Arc::new(AtomicU64::new(IggyTimestamp::now().as_micros())),
- cache_advisor_task: None,
- };
-
- reader.start_cache_advisor_task();
- Ok(reader)
+ file: Arc::new(file.into_std().await),
+ messages_size_bytes,
+ })
}
- /// Loads and returns all message IDs from the log file.
+ /// Loads and returns all message IDs from the messages file.
pub async fn load_all_message_ids_from_disk(
&self,
indexes: IggyIndexes,
@@ -116,7 +103,7 @@ impl MessagesReader {
Ok(ids)
}
- /// Loads and returns a batch of messages from the log file.
+ /// Loads and returns a batch of messages from the messages file.
pub async fn load_messages_from_disk(
&self,
indexes: IggyIndexes,
@@ -152,15 +139,13 @@ impl MessagesReader {
))
}
- /// Returns the size of the log file in bytes.
+ /// Returns the size of the messages file in bytes.
pub fn file_size(&self) -> u32 {
- self.log_size_bytes.load(Ordering::Acquire) as u32
+ self.messages_size_bytes.load(Ordering::Acquire) as u32
}
- /// Reads `len` bytes from the log file at the specified `offset`.
+ /// Reads `len` bytes from the messages file at the specified `offset`.
async fn read_bytes_at(&self, offset: u64, len: u32) -> Result<Bytes, std::io::Error> {
- self.last_access
- .store(IggyTimestamp::now().as_micros(), Ordering::Release);
let file = self.file.clone();
spawn_blocking(move || {
let mut buf = BytesMut::with_capacity(len as usize);
@@ -170,76 +155,4 @@ impl MessagesReader {
})
.await?
}
-
- /// Starts a background task that manages file caching based on access patterns
- fn start_cache_advisor_task(&mut self) {
- let file = self.file.clone();
- let file_path = self.file_path.clone();
- let last_access = self.last_access.clone();
- let log_size_bytes = self.log_size_bytes.clone();
-
- const SEQUENTIAL_ACCESS_THRESHOLD_SECS: u64 = 1;
- const DONTNEED_THRESHOLD_SECS: u64 = 5;
- const CHECK_INTERVAL_SECS: u64 = 1;
-
- let task = tokio::spawn(async move {
- let mut last_advice: Option<nix::fcntl::PosixFadviseAdvice> = None;
- let mut interval = interval(Duration::from_secs(CHECK_INTERVAL_SECS));
-
- #[cfg(not(target_os = "macos"))]
- loop {
- interval.tick().await;
-
- let now = IggyTimestamp::now().as_micros();
- let last_access_time = last_access.load(Ordering::Acquire);
- let elapsed_secs = (now - last_access_time) / 1_000_000;
-
- let fd = file.as_raw_fd();
- let file_size = log_size_bytes.load(Ordering::Acquire);
-
- let advice = if elapsed_secs < SEQUENTIAL_ACCESS_THRESHOLD_SECS {
- Some(nix::fcntl::PosixFadviseAdvice::POSIX_FADV_SEQUENTIAL)
- } else if elapsed_secs > DONTNEED_THRESHOLD_SECS {
- Some(nix::fcntl::PosixFadviseAdvice::POSIX_FADV_DONTNEED)
- } else {
- None
- };
-
- if let Some(adv) = advice {
- if last_advice != Some(adv) {
- match nix::fcntl::posix_fadvise(fd, 0, file_size as i64, adv) {
- Ok(_) => {
- if adv == nix::fcntl::PosixFadviseAdvice::POSIX_FADV_DONTNEED {
- trace!(
- "Set DONTNEED advice on file: {}, no access for {}s",
- file_path,
- elapsed_secs
- );
- } else if adv
- == nix::fcntl::PosixFadviseAdvice::POSIX_FADV_SEQUENTIAL
- {
- trace!(
- "Set SEQUENTIAL advice on file: {}, recent access {}s ago",
- file_path,
- elapsed_secs
- );
- }
- last_advice = Some(adv);
- }
- Err(e) => {
- debug!("Failed to set posix_fadvise on file {}: {}", file_path, e);
- }
- }
- }
- }
- }
-
- #[cfg(target_os = "macos")]
- loop {
- interval.tick().await;
- }
- });
-
- self.cache_advisor_task = Some(task);
- }
}
diff --git a/server/src/streaming/segments/messages/messages_writer.rs b/server/src/streaming/segments/messages/messages_writer.rs
index 0987d43c..c8b70adc 100644
--- a/server/src/streaming/segments/messages/messages_writer.rs
+++ b/server/src/streaming/segments/messages/messages_writer.rs
@@ -32,28 +32,37 @@ impl MessagesWriter {
messages_size_bytes: Arc<AtomicU64>,
fsync: bool,
server_confirmation: Confirmation,
+ file_exists: bool,
) -> Result<Self, IggyError> {
let file = OpenOptions::new()
.write(true)
- .append(true)
.create(true)
+ .append(true)
.open(file_path)
.await
.map_err(|_| IggyError::CannotReadFile)?;
- let _ = file.sync_all().await.with_error_context(|error| {
- format!("Failed to fsync messages file after creation: {file_path}. {error}",)
- });
+ if file_exists {
+ let _ = file.sync_all().await.with_error_context(|error| {
+ format!("Failed to fsync messages file after creation: {file_path}, error: {error}")
+ });
- let actual_messages_size = file
- .metadata()
- .await
- .map_err(|_| IggyError::CannotReadFileMetadata)?
- .len();
+ let actual_messages_size = file
+ .metadata()
+ .await
+ .with_error_context(|error| {
+ format!("Failed to get metadata of messages file: {file_path}, error: {error}")
+ })
+ .map_err(|_| IggyError::CannotReadFileMetadata)?
+ .len();
- messages_size_bytes.store(actual_messages_size, Ordering::Release);
+ messages_size_bytes.store(actual_messages_size, Ordering::Release);
+ }
- trace!("Opened messages file for writing: {file_path}, size: {actual_messages_size}");
+ trace!(
+ "Opened messages file for writing: {file_path}, size: {}",
+ messages_size_bytes.load(Ordering::Acquire)
+ );
let (file, persister_task) = match server_confirmation {
Confirmation::NoWait => {
diff --git a/server/src/streaming/segments/messages_accumulator.rs b/server/src/streaming/segments/messages_accumulator.rs
index 397ef73e..ef792d18 100644
--- a/server/src/streaming/segments/messages_accumulator.rs
+++ b/server/src/streaming/segments/messages_accumulator.rs
@@ -17,7 +17,7 @@ pub struct MessagesAccumulator {
/// Current (latest) offset in the accumulator
current_offset: u64,
- /// Current byte position for the next message in the segment
+ /// Current (latest) byte position for the next message in the segment
current_position: u32,
/// Collection of all message batches in the accumulator
@@ -59,7 +59,7 @@ impl MessagesAccumulator {
}
trace!(
- "Coalescing batch base_offset={}, segment_current_offset={}, messages_count={}, batch_count={}",
+ "Coalescing batch with base_offset: {}, segment_current_offset: {}, self.messages_count: {}, batch.count: {}",
self.base_offset,
segment_current_offset,
self.messages_count,
@@ -182,4 +182,9 @@ impl MessagesAccumulator {
pub fn into_batch_set(self) -> IggyMessagesBatchSet {
self.batches
}
+
+ /// Gets the size of the accumulated messages in bytes
+ pub fn size(&self) -> usize {
+ self.batches.size() as usize
+ }
}
diff --git a/server/src/streaming/segments/reading_messages.rs b/server/src/streaming/segments/reading_messages.rs
index fe893dbd..30edc01d 100644
--- a/server/src/streaming/segments/reading_messages.rs
+++ b/server/src/streaming/segments/reading_messages.rs
@@ -10,7 +10,9 @@ const COMPONENT: &str = "STREAMING_SEGMENT";
impl Segment {
pub fn get_messages_size(&self) -> IggyByteSize {
- IggyByteSize::from(self.messages_size.load(Ordering::Relaxed))
+ let on_disk_size = self.messages_size.load(Ordering::Relaxed);
+ let accumulator_size = self.accumulator.size() as u64;
+ IggyByteSize::from(on_disk_size + accumulator_size)
}
pub fn get_messages_count(&self) -> u32 {
diff --git a/server/src/streaming/segments/segment.rs b/server/src/streaming/segments/segment.rs
index b6fe7e2b..c6a5f8ae 100644
--- a/server/src/streaming/segments/segment.rs
+++ b/server/src/streaming/segments/segment.rs
@@ -106,7 +106,7 @@ impl Segment {
/// Load the segment state from disk.
pub async fn load_from_disk(&mut self) -> Result<(), IggyError> {
if self.messages_reader.is_none() || self.index_reader.is_none() {
- self.initialize_writing().await?;
+ self.initialize_writing(true).await?;
self.initialize_reading().await?;
}
@@ -181,31 +181,37 @@ impl Segment {
pub async fn persist(&mut self) -> Result<(), IggyError> {
info!("Saving segment with start offset: {} for partition with ID: {} for topic with ID: {} and stream with ID: {}",
self.start_offset, self.partition_id, self.topic_id, self.stream_id);
- self.initialize_writing().await?;
+ self.initialize_writing(false).await?;
self.initialize_reading().await?;
info!("Saved segment log file with start offset: {} for partition with ID: {} for topic with ID: {} and stream with ID: {}",
self.start_offset, self.partition_id, self.topic_id, self.stream_id);
Ok(())
}
- pub async fn initialize_writing(&mut self) -> Result<(), IggyError> {
+ pub async fn initialize_writing(&mut self, file_exists: bool) -> Result<(), IggyError> {
let log_fsync = self.config.partition.enforce_fsync;
let index_fsync = self.config.partition.enforce_fsync;
let server_confirmation = self.config.segment.server_confirmation;
- let log_writer = MessagesWriter::new(
+ let messages_writer = MessagesWriter::new(
&self.messages_path,
self.messages_size.clone(),
log_fsync,
server_confirmation,
+ file_exists,
)
.await?;
- let index_writer =
- IndexWriter::new(&self.index_path, self.indexes_size.clone(), index_fsync).await?;
+ let index_writer = IndexWriter::new(
+ &self.index_path,
+ self.indexes_size.clone(),
+ index_fsync,
+ file_exists,
+ )
+ .await?;
- self.messages_writer = Some(log_writer);
+ self.messages_writer = Some(messages_writer);
self.index_writer = Some(index_writer);
Ok(())
}
diff --git a/server/src/streaming/segments/types/messages_batch_mut.rs b/server/src/streaming/segments/types/messages_batch_mut.rs
index ad4174c3..122bbfbb 100644
--- a/server/src/streaming/segments/types/messages_batch_mut.rs
+++ b/server/src/streaming/segments/types/messages_batch_mut.rs
@@ -8,7 +8,7 @@ use iggy::prelude::*;
use iggy::utils::timestamp::IggyTimestamp;
use lending_iterator::prelude::*;
use std::ops::Deref;
-use tracing::{error, trace};
+use tracing::error;
/// A container for mutable messages that are being prepared for persistence.
///
@@ -218,15 +218,8 @@ impl IggyMessagesBatchMut {
/// Get the message at the specified index.
/// Returns None if the index is out of bounds or the message cannot be found.
pub fn get(&self, index: usize) -> Option<IggyMessageView> {
- self.get_message_boundaries(index).map(|(start, end)| {
- trace!(
- "Reading message at index {} from positions {}..{}",
- index,
- start,
- end
- );
- IggyMessageView::new(&self.messages[start..end])
- })
+ self.get_message_boundaries(index)
+ .map(|(start, end)| IggyMessageView::new(&self.messages[start..end]))
}
/// This helper function is used to parse newly appended chunks in the `new_buffer`.
diff --git a/server/src/streaming/segments/types/messages_batch_set.rs b/server/src/streaming/segments/types/messages_batch_set.rs
index 5d3d7210..7b28e2d2 100644
--- a/server/src/streaming/segments/types/messages_batch_set.rs
+++ b/server/src/streaming/segments/types/messages_batch_set.rs
@@ -59,13 +59,11 @@ impl IggyMessagesBatchSet {
self.batches.extend(other.batches);
}
- // todo usize
/// Get the total number of messages in the batch
pub fn count(&self) -> u32 {
self.count
}
- // todo usize
/// Get the total size of all messages in bytes
pub fn size(&self) -> u32 {
self.size
diff --git a/server/src/streaming/segments/writing_messages.rs b/server/src/streaming/segments/writing_messages.rs
index d43f65c1..8148eccd 100644
--- a/server/src/streaming/segments/writing_messages.rs
+++ b/server/src/streaming/segments/writing_messages.rs
@@ -58,7 +58,6 @@ impl Segment {
let accumulator = std::mem::take(&mut self.accumulator);
- // todo change to put
accumulator.update_indexes(&mut self.indexes);
let batches = accumulator.into_batch_set();
@@ -131,19 +130,25 @@ impl Segment {
.fetch_add(messages_count, Ordering::SeqCst);
self.messages_count_of_parent_partition
.fetch_add(messages_count, Ordering::SeqCst);
- self.messages_size
- .fetch_add(messages_size, Ordering::SeqCst);
}
async fn check_and_handle_segment_full(&mut self) -> Result<(), IggyError> {
if self.is_full().await {
+ let max_segment_size_from_config = self.config.segment.size;
+ let current_segment_size = self.get_messages_size();
+ assert!(
+ current_segment_size >= max_segment_size_from_config,
+ "Current segment size: {} is greater than max segment size: {}",
+ current_segment_size,
+ max_segment_size_from_config
+ );
if self.config.segment.cache_indexes == CacheIndexesConfig::OpenSegment {
self.drop_indexes();
}
self.shutdown_writing().await;
info!(
- "Closed segment with start offset: {}, end offset: {} for partition with ID: {}.",
- self.start_offset, self.end_offset, self.partition_id
+ "Closed segment with start offset: {}, end offset: {}, size: {} for partition with ID: {}.",
+ self.start_offset, self.end_offset, self.get_messages_size(), self.partition_id
);
self.is_closed = true;
}