numinnex commented on code in PR #4063:
URL: https://github.com/apache/iggy/pull/4063#discussion_r3939951642
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2421,6 +2776,30 @@ where
}
};
+ if args.auto_commit
+ && self.consensus.is_primary()
+ && let Ok(pending) =
PendingConsumerOffsetCommit::try_from_polling_consumer(consumer, 0)
+ {
+ let capacity = self.consumer_offset_capacity_for(pending.kind);
+ if !capacity.is_uncertain()
+ && self.consumer_offset_map_count(pending.kind) >=
capacity.limit()
+ {
Review Comment:
**`papaya::HashMap::len()` on every auto-commit poll.**
`len()` is `Counter::sum()` over
`available_parallelism().next_power_of_two()` `CachePadded<AtomicIsize>`
shards, so on a 64-128 core host this is 4-16 KiB of cache-line loads per poll
to compare against a limit that is almost never reached.
The `contains_key` probe at :2787-2796 is the cheaper test and is already
written — on a hit neither ordering does anything, so probing first is
semantics-preserving and free.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2355,14 +2613,111 @@ where
ReplicaLogContext::from_consensus(self.consensus(),
PlaneKind::Partitions)
}
- fn clear_pending_consumer_offset_commits_if_view_changed(&mut self) {
+ fn store_offset_range_error(&self, offset: u64) -> Option<IggyError> {
+ let current = self.stats.current_offset();
+ (offset > current || (current == 0 &&
self.stats.messages_count_inconsistent() == 0))
+ .then_some(IggyError::InvalidOffset(offset))
+ }
+
+ fn resynchronize_consumer_offset_reservations(&mut self) {
let current_view = self.consensus.view();
- if current_view == self.observed_view {
+ if current_view == self.observed_view &&
!self.offset_reservations_need_resync.get() {
return;
}
- self.pending_consumer_offset_commits.clear();
+ if current_view != self.observed_view {
+ self.queued_auto_commit_reservations.borrow_mut().clear();
+ }
+
+ let from_op = self
Review Comment:
**Scan floors at `commit_min`, but eviction is bounded by `commit_max`.**
`flush_committed_messages` evicts `committed_prefix(commit_max())` (:4516),
so ops in `(commit_min, commit_max]` lose their resident header and survive
only in the evicted ring — capped at `EVICTED_RING_CAPACITY = 4096` *or*
`EVICTED_RING_BYTES_MAX = 16 MiB`, whichever trips first, which is a handful of
batches on a produce-heavy partition. `repair_headers_in` can then legitimately
return fewer than `expected`, latching the `uncertain` flag at :2643.
This fires at promotion: a new primary runs resync from
`on_commit`/`on_replicate`/`on_request` before `VsrAction::CommitJournal` has
caught `commit_min` up. `consensus/src/impls.rs:2880-2883` states the rule for
the identical problem — "`commit_max`, not `commit_min` … `commit_min` can lag
far enough to overflow the rebuild".
Note a second trigger that flooring at `commit_max` does **not** fix:
`truncate_uncommitted_from` (:4353) drops entries without rewinding the
sequencer and arms the resync at :4436, leaving `to_op > journal head`.
Bounding `to_op` by `journal().inner.last_op()` covers both.
##########
core/server/src/partition_helpers.rs:
##########
@@ -253,15 +269,51 @@ pub fn configure_consumer_offsets(
.enforce_fsync
.unwrap_or(iggy_common::DEFAULT_ENFORCE_FSYNC);
partition.configure_consumer_offset_storage(
- consumer_offsets_path,
- consumer_group_offsets_path,
+ consumer_offsets_path.clone(),
+ consumer_group_offsets_path.clone(),
consumer_offsets,
consumer_group_offsets,
enforce_fsync,
);
+ for consumer_id in numeric_offset_file_ids(&consumer_offsets_path) {
+ if partition.seed_stranded_consumer_offset(ConsumerKind::Consumer,
consumer_id) {
+ warn!(stream_id, topic_id, partition_id, consumer_id, path =
%consumer_offsets_path,
+ "unloaded consumer offset file retains its capacity slot until
updated or deleted");
+ }
+ }
+ for group_id in numeric_offset_file_ids(&consumer_group_offsets_path) {
+ if
partition.seed_stranded_consumer_offset(ConsumerKind::ConsumerGroup, group_id) {
+ warn!(stream_id, topic_id, partition_id, group_id, path =
%consumer_group_offsets_path,
+ "unloaded group offset file retains its capacity slot until
repaired or reclaimed");
+ }
+ }
+ for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] {
+ let count = partition.occupied_consumer_offset_count(kind);
+ if count > config.partition.consumer_offsets_max {
+ warn!(
+ stream_id,
+ topic_id,
+ partition_id,
+ ?kind,
+ count,
+ limit = config.partition.consumer_offsets_max,
+ "recovered consumer offsets exceed the configured admission
limit"
+ );
+ }
+ }
Ok(())
}
+fn numeric_offset_file_ids(path: &str) -> Vec<u32> {
+ let Ok(entries) = std::fs::read_dir(path) else {
+ return Vec::new();
+ };
+ entries
+ .filter_map(Result::ok)
+ .filter_map(|entry| entry.file_name().to_str()?.parse().ok())
Review Comment:
**No `is_file()` filter, and this is a second walk of a directory already
read.**
A numerically-named *subdirectory* under the offsets dir parses fine here
and is seeded as a stranded key, permanently burning a quota slot that nothing
can clear — `clear_stranded` only fires on a successful persist or unlink for
that id. The sibling loader skips dirs (`offset_recovery.rs:132`), and this
PR's own harness helper checks `is_file()`
(`integration/src/harness/disk.rs:60`), so it is a three-way asymmetry.
It is also a second blocking `std::fs::read_dir` over the two dirs
`load_consumer_offsets` just walked, and not boot-only — `load_partition` runs
from the reconciler at runtime (`partition_reconciler.rs:734`), so it blocks a
serving shard thread.
Returning the id set (filtered) from the existing walk closes both.
Worth noting this is also the no-fault way to populate `stranded`, which is
what makes the widened delete admission at `iggy_partition.rs:2597-2601`
reachable on a mixed-version cluster.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -960,47 +994,108 @@ async fn tear_down_owned_partition(
counters.removed_local += 1;
}
-/// Reclaim consumer-group offsets left behind by a `DeleteConsumerGroup` whose
-/// topic still exists (a topic/stream delete already drops the whole partition
-/// directory, offsets included). For each owned partition, any stored
-/// consumer-group offset whose group id is no longer present in the topic's
-/// committed metadata is removed (in-memory entry + persisted file).
Monotonic,
-/// never-reused group ids make this purely reclamation -- a recreated group
-/// gets a fresh id and never reads a dead group's offset -- so it is safe to
do
-/// lazily on the reconcile pass rather than synchronously on delete.
-async fn reconcile_consumer_group_offsets(ctx: &ReconcilerCtx, counters: &mut
PassCounters) {
+/// Reclaim deleted groups through ordered offset deletes. Replicas must see
+/// each delete before a replacement store can reuse its durable slot.
+fn reconcile_consumer_group_offsets(ctx: &ReconcilerCtx, counters: &mut
PassCounters) {
let live_groups = snapshot_topic_live_groups(ctx);
let partitions = ctx.shard.plane.partitions();
let owned: Vec<IggyNamespace> = partitions.namespaces().copied().collect();
- for ns in owned {
- let live = live_groups.get(&(ns.stream_id(), ns.topic_id()));
- // Take the in-memory removes + owned unlink paths under a
closure-scoped
- // borrow that cannot escape into the await below. Holding a raw
- // `&IggyPartition` across `delete_persisted_offset().await` would let
the
- // pump task realloc the partitions vec underneath us (a UAF).
- let paths = partitions.with_partition(&ns, |partition| {
- partition.reclaim_dead_group_offsets(|group_id| {
- live.is_some_and(|set| set.contains(&group_id))
- })
- });
- let Some(paths) = paths else {
+ for namespace in owned {
+ if ctx
+ .group_offset_cleanup_inflight
+ .borrow()
+ .contains(&namespace)
+ {
continue;
- };
- for path in paths {
- if let Err(err) = delete_persisted_offset(&path).await {
- warn!(
- shard = ctx.shard.id,
- ns_raw = ns.inner(),
- error = %err,
- "reconciler failed to reclaim deleted consumer-group
offset"
- );
- continue;
+ }
+ let live = live_groups.get(&(namespace.stream_id(),
namespace.topic_id()));
Review Comment:
**`live_groups` miss conflates "no live groups" with "STM not caught up" —
and now issues replicated deletes.**
`snapshot_topic_live_groups` skips topics whose `consumer_groups` map is
empty, so a topic absent from the snapshot and a topic with zero groups are the
same `None` here. `live.is_some_and(...)` at :1015 then answers false for
*every* group, and the pass enqueues a delete for every group offset on the
partition.
Pre-PR that cost a node-local unlink other replicas could restore. Post-PR
it submits durable, quorum-committed deletes (:1024), so a shard whose metadata
STM is still replaying group creates — the boot window — can permanently
destroy every live group's committed offset cluster-wide. Nothing surfaces to
the client; consumers silently re-consume from the retention floor.
Distinguishing "topic absent from this snapshot" from "topic present with
zero groups" before any delete is submitted would close it.
##########
core/server/src/consumer_group.rs:
##########
@@ -226,15 +226,28 @@ where
let body = request_body(&request);
// The store/delete ops differ only in the decode type; this collapses
// their identical decode -> resolve group id -> rewrite consumer id ->
- // re-encode bodies. A non-group consumer or unresolved group returns the
- // request untouched (the apply/read path handles the miss).
+ // re-encode bodies. Individual consumers pass through. A group identifier
+ // that metadata cannot resolve is rejected before it can create a raw file
+ // in the group-offset directory.
macro_rules! rewrite_group_offset {
($ty:ty) => {{
let mut wire = <$ty>::decode_from(body).map_err(|_|
IggyError::InvalidCommand)?;
+ if wire.consumer.kind != KIND_CONSUMER_GROUP {
+ return Ok(request);
+ }
let Some(group_id) =
resolve_group_offset_id(shard, &wire.consumer,
(&wire.stream_id, &wire.topic_id))
else {
- return Ok(request);
+ // An unknown stream or topic is not a missing group: let the
+ // namespace resolution below answer it with the same
+ // not-found every other partition op reports.
+ if !topic_exists(shard, &wire.stream_id, &wire.topic_id) {
+ return Ok(request);
+ }
+ return Err(missing_consumer_group_error(
Review Comment:
**Wire-visible error-code change, undocumented.**
Store/delete consumer offset against a missing group returned
`ResourceNotFound` (20); this now returns `ConsumerGroupIdNotFound` (5000) or
`ConsumerGroupNameNotFound` (5003) depending on the identifier form. Unknown
stream/topic still returns 20 (:244-245), which is the discriminator.
Any client branching on 20 for this op breaks. Go is the sharpest case —
`ierror.ErrResourceNotFound` is a typed error matched with `errors.Is`, so this
reroutes control flow silently.
Only the Go BDD was updated
(`bdd/go/tests/tcp_test/offset_feature_delete.go:133`) out of seven SDKs, and
there is no changelog entry or coverage for the name form (5003).
##########
core/common/src/error/iggy_error.rs:
##########
@@ -333,6 +333,8 @@ pub enum IggyError {
NotResolvedConsumer(Identifier) = 3022,
#[error("Cannot open consumer offsets file for path: {0}")]
CannotOpenConsumerOffsetsFile(String) = 3023,
+ #[error("Too many consumer offsets for partition")]
Review Comment:
**Message names neither the limit nor the knob.**
Since 3024 is terminal in every SDK retry classifier, this string is the
operator's entire remediation surface — and it does not say what the limit is
or that `[partition] consumer_offsets_max` controls it. `config.toml` is the
only doc surface in the repo, so nothing else closes the gap.
Something like "per-partition consumer offset limit reached (see
`[partition] consumer_offsets_max`)" would make it actionable.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2355,14 +2613,111 @@ where
ReplicaLogContext::from_consensus(self.consensus(),
PlaneKind::Partitions)
}
- fn clear_pending_consumer_offset_commits_if_view_changed(&mut self) {
+ fn store_offset_range_error(&self, offset: u64) -> Option<IggyError> {
+ let current = self.stats.current_offset();
+ (offset > current || (current == 0 &&
self.stats.messages_count_inconsistent() == 0))
+ .then_some(IggyError::InvalidOffset(offset))
+ }
+
+ fn resynchronize_consumer_offset_reservations(&mut self) {
let current_view = self.consensus.view();
- if current_view == self.observed_view {
+ if current_view == self.observed_view &&
!self.offset_reservations_need_resync.get() {
return;
}
- self.pending_consumer_offset_commits.clear();
+ if current_view != self.observed_view {
+ self.queued_auto_commit_reservations.borrow_mut().clear();
+ }
+
+ let from_op = self
+ .consensus
+ .commit_min()
+ .max(self.purge_floor_op)
+ .saturating_add(1);
+ let to_op = self.consensus.sequencer().current_sequence();
+ let mut rebuilt = HashMap::new();
+ let headers =
self.log.journal().inner.repair_headers_in(from_op..=to_op);
+ let expected = to_op
+ .checked_sub(from_op)
+ .map_or(0, |span| span.saturating_add(1));
+ let mut decode_failed = headers.len() as u64 != expected;
+ for (op, header) in headers {
+ if !matches!(
+ header.operation,
+ Operation::StoreConsumerOffset |
Operation::DeleteConsumerOffset
+ ) {
+ continue;
+ }
+ match self.restage_consumer_offset_from_journal(op) {
+ Ok(pending) => {
+ rebuilt.insert(op, pending);
+ }
+ Err(error) => {
+ error!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ replica_id = self.consensus.replica(),
+ namespace_raw = self.namespace().inner(),
+ op,
+ %error,
+ "failed to rebuild consumer offset reservations after
view change"
+ );
+ decode_failed = true;
+ break;
+ }
+ }
+ }
+ self.pending_consumer_offset_commits = rebuilt;
+ if decode_failed {
+ self.consumer_offset_capacity.mark_uncertain();
+ self.consumer_group_offset_capacity.mark_uncertain();
+ } else {
+ let consumer_ids = self
+ .pending_consumer_offset_commits
+ .values()
+ .filter(|pending| {
+ pending.kind == ConsumerKind::Consumer
+ && matches!(pending.mutation,
PendingConsumerOffsetMutation::Upsert(_))
+ })
+ .map(|pending| pending.consumer_id);
+ self.consumer_offset_capacity
+ .rebuild(&self.durable_consumer_offsets, consumer_ids);
+ let group_ids = self
+ .pending_consumer_offset_commits
+ .values()
+ .filter(|pending| {
+ pending.kind == ConsumerKind::ConsumerGroup
+ && matches!(pending.mutation,
PendingConsumerOffsetMutation::Upsert(_))
+ })
+ .map(|pending| pending.consumer_id);
+ self.consumer_group_offset_capacity
+ .rebuild(&self.durable_consumer_offsets, group_ids);
+ }
self.observed_view = current_view;
+ self.consumer_group_offsets_need_reconcile.set(true);
+ // Repair and truncation rearm this flag when journal contents change.
+ // A failed decode alone must not cause a full scan on every tick.
+ self.offset_reservations_need_resync.set(false);
+ if !decode_failed && self.consensus.is_primary() {
+ self.reclaim_phantom_offsets(ConsumerKind::Consumer);
Review Comment:
**Phantom reclaim is primary-gated; phantom *creation* is not.**
Backup-served auto-commit polls create live-map entries (`poll_plan.rs`) via
the `primary == Some(Ok(false))` → `mark_served()` branch — never replicated,
never reclaimed, and both reclaim call sites (here and :2780) are behind
`is_primary()`.
A backup's map therefore saturates at `consumer_offsets_max` and never
drains. Every new consumer polling that backup then gets terminal
`TooManyConsumerOffsets` while the leader still has room, healed only by
promotion. 3024 is terminal in every first-party SDK classifier (Go
`tcp_core.go:859`, Java `:786`, Node `:1039`, .NET `:292`; Rust has no arm) —
no retry, no failover, no leader redirect.
Reclaim is node-local and touches no replicated state, so it is safe on a
backup. On a backup `provisional` is provably empty, which makes the predicate
strictly cheaper there.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2355,14 +2613,111 @@ where
ReplicaLogContext::from_consensus(self.consensus(),
PlaneKind::Partitions)
}
- fn clear_pending_consumer_offset_commits_if_view_changed(&mut self) {
+ fn store_offset_range_error(&self, offset: u64) -> Option<IggyError> {
+ let current = self.stats.current_offset();
+ (offset > current || (current == 0 &&
self.stats.messages_count_inconsistent() == 0))
+ .then_some(IggyError::InvalidOffset(offset))
+ }
+
+ fn resynchronize_consumer_offset_reservations(&mut self) {
let current_view = self.consensus.view();
- if current_view == self.observed_view {
+ if current_view == self.observed_view &&
!self.offset_reservations_need_resync.get() {
return;
}
- self.pending_consumer_offset_commits.clear();
+ if current_view != self.observed_view {
+ self.queued_auto_commit_reservations.borrow_mut().clear();
+ }
+
+ let from_op = self
+ .consensus
+ .commit_min()
+ .max(self.purge_floor_op)
+ .saturating_add(1);
+ let to_op = self.consensus.sequencer().current_sequence();
+ let mut rebuilt = HashMap::new();
+ let headers =
self.log.journal().inner.repair_headers_in(from_op..=to_op);
+ let expected = to_op
+ .checked_sub(from_op)
+ .map_or(0, |span| span.saturating_add(1));
+ let mut decode_failed = headers.len() as u64 != expected;
Review Comment:
**`uncertain` latches permanently — nothing re-arms it.**
`decode_failed` sets the latch, and this same pass then stamps
`observed_view = current_view` (:2696) and clears
`offset_reservations_need_resync` (:2700). Both are the gate conditions at
:2624, so every later call early-returns and only `rebuild()` clears
`uncertain` — reachable solely from a *next* view change, truncate, reanchor,
install or purge.
While latched, every new offset key answers `TransientNotAccepted` and every
auto-commit poll for a non-durable consumer returns zero messages. Clients
treat 58 as replayable, so each request burns a full roster walk and then
fails; admission is primary-only, so re-targeting can never succeed.
Suggest re-arming `offset_reservations_need_resync` on journal growth (or
retrying the rebuild per tick) rather than clearing it after a failed decode.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2008,54 +2110,95 @@ where
// durably stored; the in-memory update is idempotent on replay
// because we look up by (kind, id).
self.persist_consumer_offset_commit(pending).await?;
- self.apply_consumer_offset_commit(pending)?;
+ self.apply_consumer_offset_commit(pending);
self.pending_consumer_offset_commits.remove(&op);
+ self.refresh_consumer_offset_reservation(pending.kind,
pending.consumer_id);
Ok(())
}
async fn persist_consumer_offset_commit(
&self,
pending: PendingConsumerOffsetCommit,
) -> Result<(), IggyError> {
- let Some(path) = self.persisted_offset_path(pending.kind,
pending.consumer_id) else {
- return Ok(());
- };
- let key = (pending.kind, pending.consumer_id);
+ let path = self.persisted_offset_path(pending.kind,
pending.consumer_id);
+ let capacity = self.consumer_offset_capacity_for(pending.kind);
+ let creates_group = pending.kind == ConsumerKind::ConsumerGroup
+ && !self
+ .durable_consumer_offsets
+ .contains(pending.kind, pending.consumer_id);
match pending.mutation {
// A server auto-commit persists monotonically: its op offset can
// trail the durably-recorded value (disk-tier polls replicate in
// IO-completion order), so a plain overwrite would rewind the file
- // and re-deliver on restart. The `persisted_offsets` tracker keeps
+ // and re-deliver on restart. The durable offset tracker keeps
// the fold off the file: a covered offset skips the write, an
// advancing one blind-writes, and only a cold key (first commit
// after boot) reads the file once. Explicit client stores
// overwrite, so a deliberate offset reset still holds. Mirrors the
// in-memory `upsert_offset_max` vs `upsert_offset` split in the
// commit-apply.
PendingConsumerOffsetMutation::Upsert(offset) if
pending.auto_commit => {
- let tracked =
self.persisted_offsets.borrow().get(&key).copied();
- let persisted = match tracked {
- Some(high_water) if offset <= high_water => return Ok(()),
- Some(_) => {
- persist_offset(&path, offset,
self.consumer_offset_enforce_fsync).await?;
- offset
+ let tracked = self
+ .durable_consumer_offsets
+ .get(pending.kind, pending.consumer_id);
+ let persisted_high_water = match (path.as_deref(), tracked) {
+ (None, _) => offset,
+ (Some(_), Some(state))
+ if state
+ .persisted_high_water
+ .is_some_and(|high_water| offset <= high_water) =>
+ {
+ state.persisted_high_water.expect("covered persisted
value")
+ }
+ (Some(path), Some(state)) => {
+ let value = state.committed_offset.max(offset);
+ persist_offset(path, value,
self.consumer_offset_enforce_fsync).await?;
+ value
}
- None => {
- persist_offset_max(&path, offset,
self.consumer_offset_enforce_fsync)
- .await?
+ (Some(path), None) => {
+ persist_offset_max(path, offset,
self.consumer_offset_enforce_fsync).await?
}
};
- self.persisted_offsets.borrow_mut().insert(key, persisted);
+ self.durable_consumer_offsets.record_auto_commit(
+ pending.kind,
+ pending.consumer_id,
+ if tracked.is_none() {
+ persisted_high_water
+ } else {
+ offset
+ },
+ persisted_high_water,
+ );
+ capacity.clear_stranded(pending.consumer_id);
+ if creates_group {
+ self.consumer_group_offsets_need_reconcile.set(true);
+ }
Ok(())
}
PendingConsumerOffsetMutation::Upsert(offset) => {
- persist_offset(&path, offset,
self.consumer_offset_enforce_fsync).await?;
- self.persisted_offsets.borrow_mut().insert(key, offset);
+ if let Some(path) = path.as_deref() {
+ persist_offset(path, offset,
self.consumer_offset_enforce_fsync).await?;
+ }
+ self.durable_consumer_offsets.record_explicit(
+ pending.kind,
+ pending.consumer_id,
+ offset,
+ Some(offset),
+ );
+ capacity.clear_stranded(pending.consumer_id);
+ if creates_group {
+ self.consumer_group_offsets_need_reconcile.set(true);
+ }
Ok(())
}
PendingConsumerOffsetMutation::Delete => {
- delete_persisted_offset(&path).await?;
- self.persisted_offsets.borrow_mut().remove(&key);
+ if let Some(path) = path.as_deref() {
+ delete_persisted_offset(path).await?;
Review Comment:
**Committed offset delete unlinks with no parent-directory fsync.**
A crash after a quorum-acked delete resurrects the file at boot, and
`configure_consumer_offsets` re-seeds both the live map and `durable` from it —
so the cursor *and* its capacity slot come back. Nothing re-issues the delete:
the op is committed and gone, and plain consumer keys have no reaper.
The crate already does this everywhere else for the stated reason: purge at
:6263-6269 ("a crash right after the purge otherwise resurrects the offset
files at boot") and install at `state_transfer.rs:2928-2935`.
It also does not stay local — the resurrected key is re-exported through
`committed_entries`, so serving one state transfer undoes the deletion
cluster-wide, which contradicts the PR's "keeps file counts identical on all
replicas".
Worth matching the existing precedents' *batch* shape rather than fsyncing
per delete: the reconciler submits up to `GROUP_OFFSET_DELETES_PER_PASS = 32`
deletes per pass, so a per-delete dir fsync is 3-30 ms of pump stall per pass.
Marking the dir dirty and issuing one `fsync_dir` per commit-walk batch gives
the same guarantee — but the flush has to be ordered *before* that batch's
replies, or the client is told "deleted" before the unlink is durable.
##########
core/partitions/src/consumer_offset_capacity.rs:
##########
@@ -0,0 +1,530 @@
+// 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::ConsumerKind;
+use std::cell::{Cell, RefCell};
+use std::collections::{HashMap, HashSet};
+use std::rc::Rc;
+use std::sync::{Arc, Weak};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DurableOffsetState {
+ pub(crate) committed_offset: u64,
+ pub(crate) persisted_high_water: Option<u64>,
+}
+
+#[derive(Debug, Default)]
+pub struct DurableConsumerOffsets {
+ consumers: RefCell<HashMap<u32, DurableOffsetState>>,
+ groups: RefCell<HashMap<u32, DurableOffsetState>>,
+}
+
+impl DurableConsumerOffsets {
+ pub(crate) fn get(&self, kind: ConsumerKind, id: u32) ->
Option<DurableOffsetState> {
+ self.entries(kind).borrow().get(&id).copied()
+ }
+
+ pub(crate) fn contains(&self, kind: ConsumerKind, id: u32) -> bool {
+ self.entries(kind).borrow().contains_key(&id)
+ }
+
+ pub(crate) fn count(&self, kind: ConsumerKind) -> usize {
+ self.entries(kind).borrow().len()
+ }
+
+ pub(crate) fn covers(&self, kind: ConsumerKind, id: u32, offset: u64) ->
bool {
+ self.get(kind, id).is_some_and(|state| {
+ state.committed_offset >= offset
+ && state
+ .persisted_high_water
+ .is_some_and(|persisted| persisted >= offset)
+ })
+ }
+
+ pub(crate) fn record_explicit(
+ &self,
+ kind: ConsumerKind,
+ id: u32,
+ committed_offset: u64,
+ persisted_high_water: Option<u64>,
+ ) {
+ self.entries(kind).borrow_mut().insert(
+ id,
+ DurableOffsetState {
+ committed_offset,
+ persisted_high_water,
+ },
+ );
+ }
+
+ pub(crate) fn record_auto_commit(
+ &self,
+ kind: ConsumerKind,
+ id: u32,
+ committed_offset: u64,
+ persisted_high_water: u64,
+ ) {
+ let mut entries = self.entries(kind).borrow_mut();
+ let state = entries.entry(id).or_insert(DurableOffsetState {
+ committed_offset,
+ persisted_high_water: None,
+ });
+ state.committed_offset = state.committed_offset.max(committed_offset);
+ state.persisted_high_water = Some(
+ state
+ .persisted_high_water
+ .unwrap_or(0)
+ .max(persisted_high_water),
+ );
+ }
+
+ pub(crate) fn mark_persisted(&self, kind: ConsumerKind, id: u32,
high_water: u64) {
+ if let Some(state) = self.entries(kind).borrow_mut().get_mut(&id) {
+ state.persisted_high_water = Some(high_water);
+ }
+ }
+
+ pub(crate) fn remove(&self, kind: ConsumerKind, id: u32) -> bool {
+ self.entries(kind).borrow_mut().remove(&id).is_some()
+ }
+
+ pub(crate) fn clear(&self) {
+ self.consumers.borrow_mut().clear();
+ self.groups.borrow_mut().clear();
+ }
+
+ pub(crate) fn committed_entries(&self, kind: ConsumerKind) -> Vec<(u32,
u64)> {
+ self.entries(kind)
+ .borrow()
+ .iter()
+ .map(|(id, state)| (*id, state.committed_offset))
+ .collect()
+ }
+
+ const fn entries(&self, kind: ConsumerKind) -> &RefCell<HashMap<u32,
DurableOffsetState>> {
+ match kind {
+ ConsumerKind::Consumer => &self.consumers,
+ ConsumerKind::ConsumerGroup => &self.groups,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ConsumerOffsetCapacityError {
+ pub kind: ConsumerKind,
+ pub occupied: usize,
+ pub limit: usize,
+ pub first_in_episode: bool,
+ pub uncertain: bool,
+}
+
+impl From<ConsumerOffsetCapacityError> for iggy_common::IggyError {
+ fn from(error: ConsumerOffsetCapacityError) -> Self {
+ if error.uncertain {
+ Self::TransientNotAccepted
+ } else {
+ Self::TooManyConsumerOffsets
+ }
+ }
+}
+
+#[derive(Debug)]
+pub struct ConsumerOffsetCapacity {
+ kind: ConsumerKind,
+ limit: Cell<usize>,
+ pending: RefCell<HashMap<u32, usize>>,
+ provisional: RefCell<HashMap<u32, Weak<()>>>,
+ stranded: RefCell<HashSet<u32>>,
+ uncertain: Cell<bool>,
+ durable_warned: Cell<bool>,
+ map_warned: Cell<bool>,
+}
+
+impl ConsumerOffsetCapacity {
+ pub(crate) fn new(kind: ConsumerKind, limit: usize) -> Self {
+ Self {
+ kind,
+ limit: Cell::new(limit),
+ pending: RefCell::new(HashMap::new()),
+ provisional: RefCell::new(HashMap::new()),
+ stranded: RefCell::new(HashSet::new()),
+ uncertain: Cell::new(false),
+ durable_warned: Cell::new(false),
+ map_warned: Cell::new(false),
+ }
+ }
+
+ pub(crate) fn set_limit(&self, limit: usize) {
+ self.limit.set(limit);
+ }
+
+ pub(crate) const fn limit(&self) -> usize {
+ self.limit.get()
+ }
+
+ pub(crate) fn try_reserve(
+ &self,
+ id: u32,
+ durable: &DurableConsumerOffsets,
+ ) -> Result<(), ConsumerOffsetCapacityError> {
+ self.check(id, durable)?;
+ *self.pending.borrow_mut().entry(id).or_default() += 1;
+ Ok(())
+ }
+
+ pub(crate) fn check(
+ &self,
+ id: u32,
+ durable: &DurableConsumerOffsets,
+ ) -> Result<(), ConsumerOffsetCapacityError> {
+ self.rearm_if_below_limit(durable);
Review Comment:
**Admission gate degrades exactly when it engages.**
`check()` calls `rearm_if_below_limit` unconditionally. Once a partition has
denied once, `durable_warned` is set (:216), so that call reaches `occupied()`
(:356) on **every** offset store and auto-commit poll — an O(|pending| +
|provisional| + |stranded|) walk with a `Weak::strong_count` atomic per
element. The latch clears only when `occupied() < limit`, which never happens
while the table is full, so the state the feature exists to handle is the state
that makes it O(n). ~4k at the default, 262144 at the ceiling, on the shared
pump — one saturated partition degrades every partition on that shard.
`d9` compounds it: the dead-`Weak` prune at :206 sits *past* the early
return at :195-204, so a workload of already-known ids never prunes and
`provisional` fills to the distinct-id count.
Cheapest correct first step is to delete this call and drive the rearm from
the mutators that can lower occupancy (`release_reservation`,
`set_pending_count(0)`, `DurableConsumerOffsets::remove`/`clear`,
`clear_stranded`) — it removes the unconditional walk without touching the
accounting.
If an O(1) count is wanted later, note a scalar counter is unsound here:
`occupied()` is a set *union* minus `durable`, so a scalar drifts on the first
overlap, and the drift direction is a partition that silently refuses every new
key forever. It needs per-id set membership (one bit per set, count moved only
on 0↔nonzero transitions) plus a real `Drop` on the reservation — releasing at
named call sites misses the frame-dropped-by-a-closing-inbox path.
##########
core/partitions/src/consumer_offset_capacity.rs:
##########
@@ -0,0 +1,530 @@
+// 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::ConsumerKind;
+use std::cell::{Cell, RefCell};
+use std::collections::{HashMap, HashSet};
+use std::rc::Rc;
+use std::sync::{Arc, Weak};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DurableOffsetState {
+ pub(crate) committed_offset: u64,
+ pub(crate) persisted_high_water: Option<u64>,
+}
+
+#[derive(Debug, Default)]
+pub struct DurableConsumerOffsets {
+ consumers: RefCell<HashMap<u32, DurableOffsetState>>,
+ groups: RefCell<HashMap<u32, DurableOffsetState>>,
+}
+
+impl DurableConsumerOffsets {
+ pub(crate) fn get(&self, kind: ConsumerKind, id: u32) ->
Option<DurableOffsetState> {
+ self.entries(kind).borrow().get(&id).copied()
+ }
+
+ pub(crate) fn contains(&self, kind: ConsumerKind, id: u32) -> bool {
+ self.entries(kind).borrow().contains_key(&id)
+ }
+
+ pub(crate) fn count(&self, kind: ConsumerKind) -> usize {
+ self.entries(kind).borrow().len()
+ }
+
+ pub(crate) fn covers(&self, kind: ConsumerKind, id: u32, offset: u64) ->
bool {
+ self.get(kind, id).is_some_and(|state| {
+ state.committed_offset >= offset
+ && state
+ .persisted_high_water
+ .is_some_and(|persisted| persisted >= offset)
+ })
+ }
+
+ pub(crate) fn record_explicit(
+ &self,
+ kind: ConsumerKind,
+ id: u32,
+ committed_offset: u64,
+ persisted_high_water: Option<u64>,
+ ) {
+ self.entries(kind).borrow_mut().insert(
+ id,
+ DurableOffsetState {
+ committed_offset,
+ persisted_high_water,
+ },
+ );
+ }
+
+ pub(crate) fn record_auto_commit(
+ &self,
+ kind: ConsumerKind,
+ id: u32,
+ committed_offset: u64,
+ persisted_high_water: u64,
+ ) {
+ let mut entries = self.entries(kind).borrow_mut();
+ let state = entries.entry(id).or_insert(DurableOffsetState {
+ committed_offset,
+ persisted_high_water: None,
+ });
+ state.committed_offset = state.committed_offset.max(committed_offset);
+ state.persisted_high_water = Some(
+ state
+ .persisted_high_water
+ .unwrap_or(0)
+ .max(persisted_high_water),
+ );
+ }
+
+ pub(crate) fn mark_persisted(&self, kind: ConsumerKind, id: u32,
high_water: u64) {
+ if let Some(state) = self.entries(kind).borrow_mut().get_mut(&id) {
+ state.persisted_high_water = Some(high_water);
+ }
+ }
+
+ pub(crate) fn remove(&self, kind: ConsumerKind, id: u32) -> bool {
+ self.entries(kind).borrow_mut().remove(&id).is_some()
+ }
+
+ pub(crate) fn clear(&self) {
+ self.consumers.borrow_mut().clear();
+ self.groups.borrow_mut().clear();
+ }
+
+ pub(crate) fn committed_entries(&self, kind: ConsumerKind) -> Vec<(u32,
u64)> {
+ self.entries(kind)
+ .borrow()
+ .iter()
+ .map(|(id, state)| (*id, state.committed_offset))
+ .collect()
+ }
+
+ const fn entries(&self, kind: ConsumerKind) -> &RefCell<HashMap<u32,
DurableOffsetState>> {
+ match kind {
+ ConsumerKind::Consumer => &self.consumers,
+ ConsumerKind::ConsumerGroup => &self.groups,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ConsumerOffsetCapacityError {
+ pub kind: ConsumerKind,
+ pub occupied: usize,
+ pub limit: usize,
+ pub first_in_episode: bool,
+ pub uncertain: bool,
+}
+
+impl From<ConsumerOffsetCapacityError> for iggy_common::IggyError {
+ fn from(error: ConsumerOffsetCapacityError) -> Self {
+ if error.uncertain {
+ Self::TransientNotAccepted
+ } else {
+ Self::TooManyConsumerOffsets
+ }
+ }
+}
+
+#[derive(Debug)]
+pub struct ConsumerOffsetCapacity {
+ kind: ConsumerKind,
+ limit: Cell<usize>,
+ pending: RefCell<HashMap<u32, usize>>,
+ provisional: RefCell<HashMap<u32, Weak<()>>>,
+ stranded: RefCell<HashSet<u32>>,
+ uncertain: Cell<bool>,
+ durable_warned: Cell<bool>,
+ map_warned: Cell<bool>,
+}
+
+impl ConsumerOffsetCapacity {
+ pub(crate) fn new(kind: ConsumerKind, limit: usize) -> Self {
+ Self {
+ kind,
+ limit: Cell::new(limit),
+ pending: RefCell::new(HashMap::new()),
+ provisional: RefCell::new(HashMap::new()),
+ stranded: RefCell::new(HashSet::new()),
+ uncertain: Cell::new(false),
+ durable_warned: Cell::new(false),
+ map_warned: Cell::new(false),
+ }
+ }
+
+ pub(crate) fn set_limit(&self, limit: usize) {
+ self.limit.set(limit);
+ }
+
+ pub(crate) const fn limit(&self) -> usize {
+ self.limit.get()
+ }
+
+ pub(crate) fn try_reserve(
+ &self,
+ id: u32,
+ durable: &DurableConsumerOffsets,
+ ) -> Result<(), ConsumerOffsetCapacityError> {
+ self.check(id, durable)?;
+ *self.pending.borrow_mut().entry(id).or_default() += 1;
+ Ok(())
+ }
+
+ pub(crate) fn check(
+ &self,
+ id: u32,
+ durable: &DurableConsumerOffsets,
+ ) -> Result<(), ConsumerOffsetCapacityError> {
+ self.rearm_if_below_limit(durable);
+ if durable.contains(self.kind, id)
+ || self.pending.borrow().contains_key(&id)
+ || self
+ .provisional
+ .borrow()
+ .get(&id)
+ .is_some_and(|token| token.strong_count() > 0)
+ || self.stranded.borrow().contains(&id)
+ {
+ return Ok(());
+ }
+ self.provisional
+ .borrow_mut()
+ .retain(|_, token| token.strong_count() > 0);
+ let occupied = self.occupied(durable);
+ let limit = self.limit.get();
+ if self.uncertain.get() || occupied >= limit {
+ return Err(ConsumerOffsetCapacityError {
+ kind: self.kind,
+ occupied,
+ limit,
+ first_in_episode: !self.durable_warned.replace(true),
+ uncertain: self.uncertain.get(),
+ });
+ }
+ Ok(())
+ }
+
+ pub(crate) fn reserve_provisional(
+ self: &Rc<Self>,
+ id: u32,
+ durable: &Rc<DurableConsumerOffsets>,
+ ) -> Result<AutoCommitReservation, ConsumerOffsetCapacityError> {
+ self.check(id, durable)?;
+ let mut provisional = self.provisional.borrow_mut();
+ let token = provisional
+ .get(&id)
+ .and_then(Weak::upgrade)
+ .unwrap_or_else(|| {
+ let token = Arc::new(());
+ provisional.insert(id, Arc::downgrade(&token));
+ token
+ });
+ Ok(AutoCommitReservation {
+ token,
+ kind: self.kind,
+ consumer_id: id,
+ })
+ }
+
+ pub(crate) fn owns(&self, reservation: &AutoCommitReservation) -> bool {
+ reservation.kind == self.kind
+ && self
+ .provisional
+ .borrow()
+ .get(&reservation.consumer_id)
+ .is_some_and(|token| std::ptr::eq(token.as_ptr(),
Arc::as_ptr(&reservation.token)))
+ }
+
+ pub(crate) fn protects(&self, id: u32, durable: &DurableConsumerOffsets)
-> bool {
+ durable.contains(self.kind, id)
+ || self.pending.borrow().contains_key(&id)
+ || self
+ .provisional
+ .borrow()
+ .get(&id)
+ .is_some_and(|token| token.strong_count() > 0)
+ }
+
+ pub(crate) fn set_pending_count(&self, id: u32, count: usize) {
+ if count == 0 {
+ self.pending.borrow_mut().remove(&id);
+ } else {
+ self.pending.borrow_mut().insert(id, count);
+ }
+ }
+
+ pub(crate) fn release_reservation(&self, id: u32) {
+ let mut pending = self.pending.borrow_mut();
+ let Some(count) = pending.get_mut(&id) else {
+ return;
+ };
+ if *count == 1 {
+ pending.remove(&id);
+ } else {
+ *count -= 1;
+ }
+ }
+
+ pub(crate) const fn is_uncertain(&self) -> bool {
+ self.uncertain.get()
+ }
+
+ pub(crate) fn rebuild(
+ &self,
+ durable: &DurableConsumerOffsets,
+ pending_ids: impl IntoIterator<Item = u32>,
+ ) {
+ let mut pending = self.pending.borrow_mut();
+ pending.clear();
+ for id in pending_ids {
+ *pending.entry(id).or_default() += 1;
+ }
+ drop(pending);
+ self.uncertain.set(false);
+ self.rearm_if_below_limit(durable);
+ }
+
+ pub(crate) fn mark_uncertain(&self) {
+ self.pending.borrow_mut().clear();
+ self.uncertain.set(true);
+ }
+
+ pub(crate) fn record_stranded(&self, id: u32) {
+ self.stranded.borrow_mut().insert(id);
+ }
+
+ pub(crate) fn clear_stranded(&self, id: u32) {
+ self.stranded.borrow_mut().remove(&id);
+ }
+
+ pub(crate) fn stranded_ids(&self) -> Vec<u32> {
+ self.stranded.borrow().iter().copied().collect()
+ }
+
+ pub(crate) fn is_stranded(&self, id: u32) -> bool {
+ self.stranded.borrow().contains(&id)
+ }
+
+ pub(crate) fn rearm_if_below_limit(&self, durable:
&DurableConsumerOffsets) {
+ if !self.durable_warned.get() || self.uncertain.get() {
+ return;
+ }
+ if self.occupied(durable) < self.limit.get() {
+ self.durable_warned.set(false);
+ }
+ }
+
+ pub(crate) const fn admit_local_map_key(
Review Comment:
**Map-full denial is terminal, but the condition is transient.**
`uncertain: false` here makes this `TooManyConsumerOffsets`, which every SDK
classifier treats as non-retryable. The condition is not terminal though: the
phantoms are reclaimable, and pre-reclaim runs at plan build while
`apply_local_offset` runs later on the detached disk task — requests landing in
that window get a permanent code for a condition that would have cleared.
Worth pairing with the SDK backoff: `sdk/src/clients/consumer.rs:1358-1366`
sleeps only for `Disconnected | Unauthenticated | StaleClient`, so
reclassifying this to 58 alone would relocate the hot-loop rather than fix it —
and 58 additionally triggers a full-roster walk (`leader_aware.rs:352` marks it
"safe to re-issue anywhere"), turning one tight loop into an N× fan-out. The
SDK change wants to land first.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2754,6 +2861,12 @@ where
entry.offset.store(value, Ordering::Release);
let path = entry.path.clone();
self.consumer_offsets.pin().insert(*id as usize, entry);
+ self.durable_consumer_offsets.record_explicit(
Review Comment:
**Durable membership is recorded before the files land.**
`record_explicit(..., None)` runs here, but `mark_persisted` only runs for
writes that succeeded (:2911). On the degraded `offsets_written == false` path
the partition holds durable membership for keys with no file on disk.
The safe half works — `covers()` is false for `None`, so the next
auto-commit blind-writes and repairs. The unsafe half: a key that never commits
again is still counted by `occupied()` and still exported by
`committed_entries`, so this node advertises in its own transfer artifact an
offset it will lose on its own restart.
The guard at :1998-2015 tests *live-map* membership rather than file
durability, so it cannot catch this.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2421,6 +2776,30 @@ where
}
};
+ if args.auto_commit
+ && self.consensus.is_primary()
+ && let Ok(pending) =
PendingConsumerOffsetCommit::try_from_polling_consumer(consumer, 0)
+ {
+ let capacity = self.consumer_offset_capacity_for(pending.kind);
+ if !capacity.is_uncertain()
+ && self.consumer_offset_map_count(pending.kind) >=
capacity.limit()
+ {
+ let exists = match pending.kind {
+ ConsumerKind::Consumer => self
+ .consumer_offsets
+ .pin()
+ .contains_key(&(pending.consumer_id as usize)),
+ ConsumerKind::ConsumerGroup => self
+ .consumer_group_offsets
+ .pin()
+ .contains_key(&ConsumerGroupId(pending.consumer_id as
usize)),
+ };
+ if !exists {
Review Comment:
**Unmemoized O(limit) sweep on the poll path, and the SDK closes the loop.**
`reclaim_phantom_offsets` is a full papaya `retain` whose predicate does up
to 3 hash lookups plus an atomic per entry. The guard above (`map_count >=
limit && !exists`) is exactly the saturated steady state, so while the map is
full every poll from a non-member repeats the whole scan — on the pump that
serves every partition on the shard.
It self-amplifies: refusal → `sdk/src/clients/consumer.rs:1358-1366` backs
off only for `Disconnected | Unauthenticated | StaleClient`, so 3024/58 fall
through with no sleep → immediate re-poll under `without_poll_interval()` →
this sweep again. Any client with poll rights and fresh consumer ids arms it;
no rate limit anywhere in the loop.
A `Cell<u64>` epoch bumped only by protection-losing events
(`durable.remove`, `release_reservation`, `set_pending_count`,
`rollback_created`, view change), skipping the `retain` when nothing changed
since the last sweep that reclaimed nothing, makes this O(1) amortized — and
cheap enough to also run on backups, which is what :2702 needs.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
