You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "numinnex (via GitHub)" <gi...@apache.org> on 2026/06/25 09:06:04 UTC
[PR] feat(partitions): implement partition background tasks [iggy]
numinnex opened a new pull request, #3564:
URL: https://github.com/apache/iggy/pull/3564
This PR adds support for background tasks for partitions (segment cleaners). Additionally it adds support for `DeleteSegment` and `PurgeTopic`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): implement partition background tasks [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #3564:
URL: https://github.com/apache/iggy/pull/3564#discussion_r3480929314
##########
core/metadata/src/stm/stream.rs:
##########
Review Comment:
The comment is over-exaggerated, we don't need to hold such levels of backwards compatibility, in fact we don't need any backwards compatibility at all (the server-ng binary isn't used anywhere, it's not released at all).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): implement partition background tasks [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #3564:
URL: https://github.com/apache/iggy/pull/3564#discussion_r3480998952
##########
core/metadata/src/stm/stream.rs:
##########
Review Comment:
I've removed the comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): implement partition background tasks [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #3564:
URL: https://github.com/apache/iggy/pull/3564#issuecomment-4844310503
@atharvalade I've addressed all of your comments, if you don't have more comments lets procede with merging this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): implement partition background tasks [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #3564:
URL: https://github.com/apache/iggy/pull/3564#issuecomment-4830251374
/ready
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): implement partition background tasks [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3564:
URL: https://github.com/apache/iggy/pull/3564#issuecomment-4797626939
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3564?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 73.24%. Comparing base ([`bc68827`](https://app.codecov.io/gh/apache/iggy/commit/bc68827aea2d844069974b6514d8f89ff0e07bf7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`8717caf`](https://app.codecov.io/gh/apache/iggy/commit/8717caf4176b09724be8baa8bc9728c600577710?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3564 +/- ##
============================================
- Coverage 74.02% 73.24% -0.78%
Complexity 937 937
============================================
Files 1247 1184 -63
Lines 127567 122873 -4694
Branches 103435 103341 -94
============================================
- Hits 94427 90004 -4423
+ Misses 30103 29832 -271
Partials 3037 3037
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/3564/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/3564/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `74.67% <ø> (+<0.01%)` | :arrow_up: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3564/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.44% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3564/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.10% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3564/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `88.88% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/3564/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.29% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3564/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `87.27% <ø> (-3.86%)` | :arrow_down: |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3564/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `40.14% <ø> (ø)` | |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3564?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/binary\_protocol/src/consensus/operation.rs](https://app.codecov.io/gh/apache/iggy/pull/3564?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fconsensus%2Foperation.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL2NvbnNlbnN1cy9vcGVyYXRpb24ucnM=) | `96.33% <ø> (-0.10%)` | :arrow_down: |
| [core/binary\_protocol/src/dispatch.rs](https://app.codecov.io/gh/apache/iggy/pull/3564?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fdispatch.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL2Rpc3BhdGNoLnJz) | `91.63% <ø> (ø)` | |
| [core/consensus/src/observability.rs](https://app.codecov.io/gh/apache/iggy/pull/3564?src=pr&el=tree&filepath=core%2Fconsensus%2Fsrc%2Fobservability.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25zZW5zdXMvc3JjL29ic2VydmFiaWxpdHkucnM=) | `41.09% <ø> (ø)` | |
| [core/metadata/src/impls/metadata.rs](https://app.codecov.io/gh/apache/iggy/pull/3564?src=pr&el=tree&filepath=core%2Fmetadata%2Fsrc%2Fimpls%2Fmetadata.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXRhZGF0YS9zcmMvaW1wbHMvbWV0YWRhdGEucnM=) | `35.52% <ø> (-0.07%)` | :arrow_down: |
| [core/metadata/src/stm/snapshot.rs](https://app.codecov.io/gh/apache/iggy/pull/3564?src=pr&el=tree&filepath=core%2Fmetadata%2Fsrc%2Fstm%2Fsnapshot.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXRhZGF0YS9zcmMvc3RtL3NuYXBzaG90LnJz) | `85.89% <ø> (ø)` | |
| [core/metadata/src/stm/stream.rs](https://app.codecov.io/gh/apache/iggy/pull/3564?src=pr&el=tree&filepath=core%2Fmetadata%2Fsrc%2Fstm%2Fstream.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXRhZGF0YS9zcmMvc3RtL3N0cmVhbS5ycw==) | `54.50% <ø> (+0.10%)` | :arrow_up: |
| [core/partitions/src/iggy\_partition.rs](https://app.codecov.io/gh/apache/iggy/pull/3564?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Figgy_partition.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9pZ2d5X3BhcnRpdGlvbi5ycw==) | `41.32% <ø> (ø)` | |
| [core/partitions/src/segment.rs](https://app.codecov.io/gh/apache/iggy/pull/3564?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Fsegment.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9zZWdtZW50LnJz) | `57.14% <ø> (+1.97%)` | :arrow_up: |
| [core/server-ng/src/bootstrap.rs](https://app.codecov.io/gh/apache/iggy/pull/3564?src=pr&el=tree&filepath=core%2Fserver-ng%2Fsrc%2Fbootstrap.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXItbmcvc3JjL2Jvb3RzdHJhcC5ycw==) | `7.30% <ø> (ø)` | |
| [core/server-ng/src/dispatch.rs](https://app.codecov.io/gh/apache/iggy/pull/3564?src=pr&el=tree&filepath=core%2Fserver-ng%2Fsrc%2Fdispatch.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXItbmcvc3JjL2Rpc3BhdGNoLnJz) | `0.00% <ø> (ø)` | |
| ... and [3 more](https://app.codecov.io/gh/apache/iggy/pull/3564?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [81 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/3564/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): implement partition background tasks [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex merged PR #3564:
URL: https://github.com/apache/iggy/pull/3564
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): implement partition background tasks [iggy]
Posted by "atharvalade (via GitHub)" <gi...@apache.org>.
atharvalade commented on code in PR #3564:
URL: https://github.com/apache/iggy/pull/3564#discussion_r3478301520
##########
core/metadata/src/stm/stream.rs:
##########
Review Comment:
Isn't this a snapshot backwards-incompatibility? The comment says "a shorter array" fills trailing defaults, but on master `round_robin_counter: usize` sat at position 9 in the positional msgpack array. Old snapshots still carry that integer at position 9, and `rmp_serde` will fail trying to deserialize it as `Vec<(u64, ConsumerGroupSnapshot)>`. If server-ng has never cut a real snapshot this is fine, but the comment's compatibility claim doesn't hold for any existing snapshot file.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2344,6 +2355,341 @@ where
Ok(())
}
+ /// Minimum committed offset across all consumers and consumer groups, with
+ /// the holder's identity. `None` when nothing has been committed, in which
+ /// case there is no deletion barrier.
+ fn min_committed_offset(&self) -> Option<(u64, ConsumerKind, u32)> {
+ let consumer_guard = self.consumer_offsets.pin();
+ let group_guard = self.consumer_group_offsets.pin();
+ let consumers = consumer_guard.iter().map(|(_, offset)| {
+ (
+ offset.offset.load(Ordering::Relaxed),
+ offset.kind,
+ offset.consumer_id,
+ )
+ });
+ let groups = group_guard.iter().map(|(_, offset)| {
+ (
+ offset.offset.load(Ordering::Relaxed),
+ offset.kind,
+ offset.consumer_id,
+ )
+ });
+ consumers.chain(groups).min_by_key(|(offset, _, _)| *offset)
+ }
+
+ /// Time-expiry plus size-retention in one pass: remove the leading sealed
+ /// segments that have expired or that push the partition past `max_bytes`.
+ /// Returns the `(segments, messages)` removed.
+ pub async fn clean_expired_segments(
+ &mut self,
+ now: IggyTimestamp,
+ message_expiry: IggyExpiry,
+ max_bytes: Option<u64>,
+ ) -> (u64, u64) {
+ let expired = leading_expired_end(self.log.segments(), now, message_expiry);
+ let oversized =
+ max_bytes.and_then(|max_bytes| leading_oversized_end(self.log.segments(), max_bytes));
+ let Some(up_to) = expired.into_iter().chain(oversized).max() else {
+ return (0, 0);
+ };
+ self.remove_sealed_segments_up_to(up_to).await
+ }
+
+ /// Remove the oldest sealed segments whose `end_offset <= up_to_offset`,
+ /// never the active segment and never past the consumer barrier (the
+ /// minimum committed consumer/group offset). Unlinks the messages and
+ /// index files and decrements partition stats. Idempotent: an offset below
+ /// the oldest sealed segment removes nothing. Returns the
+ /// `(segments, messages)` removed.
+ ///
+ /// Holds `write_lock` to serialize against the commit/rotate path, which
+ /// runs on the separate consensus-tick loop.
+ pub async fn remove_sealed_segments_up_to(&mut self, up_to_offset: u64) -> (u64, u64) {
+ let write_lock = self.write_lock.clone();
+ let _guard = write_lock.lock().await;
+
+ let barrier = self.min_committed_offset();
+ let namespace = self.namespace();
+ let removable = {
+ let segments = self.log.segments();
+ let last_idx = segments.len().saturating_sub(1);
+ let mut removable = 0usize;
+ for (idx, segment) in segments.iter().enumerate() {
+ if idx == last_idx || !segment.sealed || segment.end_offset > up_to_offset {
+ break;
+ }
+ if let Some((barrier_offset, kind, consumer_id)) = barrier
+ && segment.end_offset > barrier_offset
+ {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = namespace.inner(),
+ start_offset = segment.start_offset,
+ end_offset = segment.end_offset,
+ barrier = barrier_offset,
+ %kind,
+ consumer_id,
+ "segment retained: blocked by committed consumer offset"
+ );
+ break;
+ }
+ removable += 1;
+ }
+ removable
+ };
+
+ let mut deleted_segments = 0u64;
+ let mut deleted_messages = 0u64;
+ for _ in 0..removable {
+ // The removable run is always a prefix (oldest first), so the next
+ // victim is index 0 once the previous one is gone.
+ let segment = self.log.segments_mut().remove(0);
+ let mut storage = self.log.storages_mut().remove(0);
+ self.log.indexes_mut().remove(0);
+ self.log.messages_writers_mut().remove(0);
+ self.log.index_writers_mut().remove(0);
+
+ let (messages_path, index_path) = storage.segment_and_index_paths();
+ let _ = storage.shutdown();
+ drop(storage);
+
+ for path in messages_path.into_iter().chain(index_path) {
+ match compio::fs::remove_file(&path).await {
+ Ok(()) => {}
+ Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+ Err(error) => {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = namespace.inner(),
+ path = %path,
+ %error,
+ "failed to unlink segment file during cleanup"
+ );
+ }
+ }
+ }
+
+ let segment_size = segment.size.as_bytes_u64();
+ let messages_in_segment = if segment.start_offset == segment.end_offset {
+ 0
+ } else {
+ segment.end_offset - segment.start_offset + 1
+ };
Review Comment:
This undercounts by 1 for sealed segments that contain exactly one message. A sealed segment with one message at offset N has `start_offset == end_offset == N`, but this branch treats it as 0 messages. Since the removal loop already guarantees `segment.sealed == true` (line 2419), a sealed segment can't be empty, so the formula should always be `end_offset - start_offset + 1`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2344,6 +2355,341 @@ where
Ok(())
}
+ /// Minimum committed offset across all consumers and consumer groups, with
+ /// the holder's identity. `None` when nothing has been committed, in which
+ /// case there is no deletion barrier.
+ fn min_committed_offset(&self) -> Option<(u64, ConsumerKind, u32)> {
+ let consumer_guard = self.consumer_offsets.pin();
+ let group_guard = self.consumer_group_offsets.pin();
+ let consumers = consumer_guard.iter().map(|(_, offset)| {
+ (
+ offset.offset.load(Ordering::Relaxed),
+ offset.kind,
+ offset.consumer_id,
+ )
+ });
+ let groups = group_guard.iter().map(|(_, offset)| {
+ (
+ offset.offset.load(Ordering::Relaxed),
+ offset.kind,
+ offset.consumer_id,
+ )
+ });
+ consumers.chain(groups).min_by_key(|(offset, _, _)| *offset)
+ }
+
+ /// Time-expiry plus size-retention in one pass: remove the leading sealed
+ /// segments that have expired or that push the partition past `max_bytes`.
+ /// Returns the `(segments, messages)` removed.
+ pub async fn clean_expired_segments(
+ &mut self,
+ now: IggyTimestamp,
+ message_expiry: IggyExpiry,
+ max_bytes: Option<u64>,
+ ) -> (u64, u64) {
+ let expired = leading_expired_end(self.log.segments(), now, message_expiry);
+ let oversized =
+ max_bytes.and_then(|max_bytes| leading_oversized_end(self.log.segments(), max_bytes));
+ let Some(up_to) = expired.into_iter().chain(oversized).max() else {
+ return (0, 0);
+ };
+ self.remove_sealed_segments_up_to(up_to).await
+ }
+
+ /// Remove the oldest sealed segments whose `end_offset <= up_to_offset`,
+ /// never the active segment and never past the consumer barrier (the
+ /// minimum committed consumer/group offset). Unlinks the messages and
+ /// index files and decrements partition stats. Idempotent: an offset below
+ /// the oldest sealed segment removes nothing. Returns the
+ /// `(segments, messages)` removed.
+ ///
+ /// Holds `write_lock` to serialize against the commit/rotate path, which
+ /// runs on the separate consensus-tick loop.
+ pub async fn remove_sealed_segments_up_to(&mut self, up_to_offset: u64) -> (u64, u64) {
+ let write_lock = self.write_lock.clone();
+ let _guard = write_lock.lock().await;
+
+ let barrier = self.min_committed_offset();
+ let namespace = self.namespace();
+ let removable = {
+ let segments = self.log.segments();
+ let last_idx = segments.len().saturating_sub(1);
+ let mut removable = 0usize;
+ for (idx, segment) in segments.iter().enumerate() {
+ if idx == last_idx || !segment.sealed || segment.end_offset > up_to_offset {
+ break;
+ }
+ if let Some((barrier_offset, kind, consumer_id)) = barrier
+ && segment.end_offset > barrier_offset
+ {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = namespace.inner(),
+ start_offset = segment.start_offset,
+ end_offset = segment.end_offset,
+ barrier = barrier_offset,
+ %kind,
+ consumer_id,
+ "segment retained: blocked by committed consumer offset"
+ );
+ break;
+ }
+ removable += 1;
+ }
+ removable
+ };
+
+ let mut deleted_segments = 0u64;
+ let mut deleted_messages = 0u64;
+ for _ in 0..removable {
+ // The removable run is always a prefix (oldest first), so the next
+ // victim is index 0 once the previous one is gone.
+ let segment = self.log.segments_mut().remove(0);
+ let mut storage = self.log.storages_mut().remove(0);
+ self.log.indexes_mut().remove(0);
+ self.log.messages_writers_mut().remove(0);
+ self.log.index_writers_mut().remove(0);
+
+ let (messages_path, index_path) = storage.segment_and_index_paths();
+ let _ = storage.shutdown();
+ drop(storage);
+
+ for path in messages_path.into_iter().chain(index_path) {
+ match compio::fs::remove_file(&path).await {
+ Ok(()) => {}
+ Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+ Err(error) => {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = namespace.inner(),
+ path = %path,
+ %error,
+ "failed to unlink segment file during cleanup"
+ );
+ }
+ }
+ }
+
+ let segment_size = segment.size.as_bytes_u64();
+ let messages_in_segment = if segment.start_offset == segment.end_offset {
+ 0
+ } else {
+ segment.end_offset - segment.start_offset + 1
+ };
+ self.stats.decrement_size_bytes(segment_size);
+ self.stats.decrement_segments_count(1);
+ self.stats.decrement_messages_count(messages_in_segment);
+
+ deleted_segments += 1;
+ deleted_messages += messages_in_segment;
+
+ debug!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = namespace.inner(),
+ start_offset = segment.start_offset,
+ end_offset = segment.end_offset,
+ "deleted sealed segment during cleanup"
+ );
+ }
+
+ (deleted_segments, deleted_messages)
+ }
+
+ /// Build and install a fresh empty segment starting at `start_offset` with
+ /// real on-disk writers. Paths are derived from the partition directory
+ /// (see `rotate_segment`); falls back to the config-derived path for
+ /// in-memory partitions with no directory.
+ ///
+ /// # Errors
+ /// If the segment's log / index file cannot be created.
+ async fn install_empty_segment(
+ &mut self,
+ config: &PartitionsConfig,
+ start_offset: u64,
+ ) -> Result<(), IggyError> {
+ let namespace = self.namespace();
+ let (messages_path, index_path) = self.partition_dir().map_or_else(
+ || {
+ (
+ config.get_messages_path(
+ namespace.stream_id(),
+ namespace.topic_id(),
+ namespace.partition_id(),
+ start_offset,
+ ),
+ config.get_index_path(
+ namespace.stream_id(),
+ namespace.topic_id(),
+ namespace.partition_id(),
+ start_offset,
+ ),
+ )
+ },
+ |dir| {
+ (
+ format!("{dir}/{start_offset:0>20}.log"),
+ format!("{dir}/{start_offset:0>20}.index"),
+ )
+ },
+ );
+ let segment = Segment::new(start_offset, config.segment_size);
+ let storage = SegmentStorage::new(
+ &messages_path,
+ &index_path,
+ 0,
+ 0,
+ config.enforce_fsync,
+ config.enforce_fsync,
+ false,
+ )
+ .await
+ .map_err(|_| IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?;
+ let messages_size_bytes = storage
+ .messages_writer
+ .as_ref()
+ .ok_or_else(|| IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?
+ .size_counter();
+ let messages_writer = Rc::new(
+ MessagesWriter::new(
+ &messages_path,
+ messages_size_bytes,
+ config.enforce_fsync,
+ false,
+ )
+ .await
+ .map_err(|_| IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?,
+ );
+ let index_writer = Rc::new(
+ IggyIndexWriter::new(
+ &index_path,
+ Rc::new(std::sync::atomic::AtomicU64::new(0)),
+ config.enforce_fsync,
+ false,
+ )
+ .await
+ .map_err(|_| IggyError::CannotCreateSegmentIndexFile(index_path.clone()))?,
+ );
+ self.log
+ .add_persisted_segment(segment, storage, Some(messages_writer), Some(index_writer));
+ Ok(())
+ }
+
+ /// Reset the partition to a single empty segment at offset 0 and clear all
+ /// consumer / consumer-group offsets (memory + disk). This is the local
+ /// effect of a committed `PurgeTopic`: it wipes message data and offsets but
+ /// preserves the partition and its consumer-group membership. Mirrors the
+ /// legacy server's `purge_all_segments` + offset-file deletion.
+ ///
+ /// Records `generation` as the applied purge generation so the reconciler
+ /// does not re-wipe a partition already purged at this generation (a later
+ /// `PurgeTopic` advances the committed generation and triggers a fresh pass).
+ ///
+ /// # Errors
+ /// If the replacement segment's log / index file cannot be created.
+ pub async fn purge(
+ &mut self,
+ config: &PartitionsConfig,
+ generation: u64,
+ ) -> Result<(), IggyError> {
+ let write_lock = self.write_lock.clone();
+ let _guard = write_lock.lock().await;
+
+ let namespace = self.namespace();
+
+ // Drain every segment (including the active one) and unlink its files.
+ let segment_count = self.log.segments().len();
+ for _ in 0..segment_count {
+ self.log.segments_mut().remove(0);
+ let mut storage = self.log.storages_mut().remove(0);
+ self.log.indexes_mut().remove(0);
+ self.log.messages_writers_mut().remove(0);
+ self.log.index_writers_mut().remove(0);
+
+ let (messages_path, index_path) = storage.segment_and_index_paths();
+ let _ = storage.shutdown();
+ drop(storage);
+
+ for path in messages_path.into_iter().chain(index_path) {
+ match compio::fs::remove_file(&path).await {
+ Ok(()) => {}
+ Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+ Err(error) => {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = namespace.inner(),
+ path = %path,
+ %error,
+ "failed to unlink segment file during purge"
+ );
+ }
+ }
+ }
+ }
+
+ // Recreate a fresh empty segment at offset 0 with real writers.
+ let start_offset = 0u64;
+ self.install_empty_segment(config, start_offset).await?;
+
+ // Reset the offset counters so new messages start at offset 0.
+ self.offset.store(start_offset, Ordering::Release);
+ self.dirty_offset.store(start_offset, Ordering::Relaxed);
+ self.should_increment_offset = false;
+
+ // Clear consumer + consumer-group offsets (memory + disk). Collect the
+ // file paths before deleting so the map guard is not held across an
+ // await.
+ let consumer_paths: Vec<String> = {
+ let guard = self.consumer_offsets.pin();
+ let paths = guard
+ .iter()
+ .filter_map(|(key, _)| {
+ u32::try_from(*key)
+ .ok()
+ .and_then(|id| self.persisted_offset_path(ConsumerKind::Consumer, id))
+ })
+ .collect();
+ guard.clear();
+ paths
+ };
+ let group_paths: Vec<String> = {
+ let guard = self.consumer_group_offsets.pin();
+ let paths = guard
+ .iter()
+ .filter_map(|(key, _)| {
+ u32::try_from(key.0)
+ .ok()
+ .and_then(|id| self.persisted_offset_path(ConsumerKind::ConsumerGroup, id))
+ })
+ .collect();
+ guard.clear();
+ paths
Review Comment:
`last_polled_offsets` is not cleared during purge. After purge resets everything to offset 0, stale `last_polled` entries survive. The reconciler's completable check (`committed >= last_polled`) then sees `None >= Some(old_high_offset)` and can't complete pending revocations until timeout. I think you need `self.last_polled_offsets.pin().clear();` alongside the other two clears.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): implement partition background tasks [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #3564:
URL: https://github.com/apache/iggy/pull/3564#discussion_r3480929314
##########
core/metadata/src/stm/stream.rs:
##########
Review Comment:
The comment is over-exaggerated, we don't need to hold to such levels of backwards compatibility, in fact we don't need any backwards compatibility at all (the server-ng binary isn't used anywhere, it's not released at all).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): implement partition background tasks [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #3564:
URL: https://github.com/apache/iggy/pull/3564#issuecomment-4843807417
/ready
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): implement partition background tasks [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #3564:
URL: https://github.com/apache/iggy/pull/3564#discussion_r3480644735
##########
core/metadata/src/stm/stream.rs:
##########
Review Comment:
Thanks — you're right that the comment overclaims, and I want to separate two things:
This comment block and the `round_robin_counter` ordering it refers to come from #3527 (the consumer-group port), not this PR. The only snapshot change here is `purge_generation`, appended as the last `#[serde(default)]` field of `PartitionSnapshot` — exactly the safe trailing-append case the comment prescribes, so this PR doesn't introduce a new compatibility hazard.
On the underlying concern itself: agreed — an older snapshot that still carried `round_robin_counter` mid-array would break `rmp_serde` (it would hit the integer where a `Vec` is now expected); `serde(default)` only rescues *trailing* gaps, not removed/reordered positions. But server-ng has no shipped or versioned snapshot format yet, so there are no such files to migrate. The #3527 comment is worth tightening to not claim general compatibility, but that's outside this PR's scope.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): implement partition background tasks [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #3564:
URL: https://github.com/apache/iggy/pull/3564#discussion_r3480644735
##########
core/metadata/src/stm/stream.rs:
##########
Review Comment:
Thanks — you're right that the comment overclaims, and I want to separate two things:
This comment block and the `round_robin_counter` ordering it refers to come from #3527 (the consumer-group port), not this PR. The only snapshot change here is `purge_generation`, appended as the last `#[serde(default)]` field of `PartitionSnapshot` — exactly the safe trailing-append case the comment prescribes, so this PR doesn't introduce a new compatibility hazard.
On the underlying concern itself: agreed — an older snapshot that still carried `round_robin_counter` mid-array would break `rmp_serde` (it would hit the integer where a `Vec` is now expected); `serde(default)` only rescues *trailing* gaps, not removed/reordered positions. But server-ng has no shipped or versioned snapshot format yet, so there are no such files to migrate. The #3527 comment is worth tightening to not claim general compatibility, but that's outside this PR's scope.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org