hubcio commented on code in PR #4063:
URL: https://github.com/apache/iggy/pull/4063#discussion_r3942247018
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -1324,7 +1324,13 @@ impl TcpClient {
// never reach the rest of the roster.
(next, true)
} else {
- return Err(IggyError::TransientNotAccepted);
+ // A one-node roster has nowhere else to walk while a
+ // freshly committed partition is still materialising.
+ // The server explicitly did not admit this request, so
+ // keep retrying the current endpoint within the
existing
+ // overall deadline rather than surfacing a transient
+ // solely because the roster contains no alternative.
+ (current, false)
Review Comment:
warning: `RosterWalk::next()` returns `None` for any exhausted walk, not
just a one-node roster, so on a 3-node cluster where every peer refuses we stop
surfacing `TransientNotAccepted` and re-hammer the last endpoint. gate on
roster size.
##########
core/sdk/src/websocket/websocket_client.rs:
##########
@@ -211,7 +211,10 @@ impl BinaryTransport for WebSocketClient {
} else if let Some(next) =
roster_walk.as_mut().and_then(RosterWalk::next) {
(next, true)
} else {
Review Comment:
warning: same as the tcp client - this fires on any exhausted roster walk,
not just a one-node roster, so a full-cluster refusal replays one endpoint
instead of surfacing `TransientNotAccepted`. gate on roster size.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2686,130 +2917,96 @@ where
"consumer group offset key {} exceeds u32",
key.0
);
- narrowed
- .and_then(|id|
self.persisted_offset_path(ConsumerKind::ConsumerGroup, id))
+ narrowed.and_then(|id| {
+
self.persisted_offset_path(ConsumerKind::ConsumerGroup, id)
+ .map(|path| (ConsumerKind::ConsumerGroup, id,
path))
+ })
})
.collect();
guard.clear();
- paths.extend(strayed_offset_files(
- self.consumer_group_offsets_path.as_deref(),
- &offsets_wire.groups,
- ));
+ paths.extend(
+ strayed_offset_files(
+ self.consumer_group_offsets_path.as_deref(),
+ &offsets_wire.groups,
+ )
+ .into_iter()
+ .filter_map(|path| {
+ numeric_offset_id(&path).map(|id|
(ConsumerKind::ConsumerGroup, id, path))
+ }),
+ );
paths
};
- for path in old_consumer_paths.into_iter().chain(old_group_paths) {
- if let Err(error) = delete_persisted_offset(&path).await {
- // Not fatal, but not silent either: a stranded file is an id
- // absent from the NEW table (matching ids get overwritten at
- // the same path), and boot resurrects it. Sharpest after a
- // purged origin ships `next_offset = 0`, where the clamp drops
- // every incoming entry and the whole old table survives while
- // the install still reports success.
- tracing::warn!(
- target: "iggy.partitions.diag",
- plane = "partitions",
- namespace_raw = self.consensus().group(),
- path = %path,
- %error,
- "failed to unlink a superseded consumer-offset file during
install"
- );
+ let mut offset_dirs_changed = [false; 2];
+ for (kind, consumer_id, path) in
old_consumer_paths.into_iter().chain(old_group_paths) {
+ let removed = delete_persisted_offset(&path)
Review Comment:
warning: an unlinkable offset file makes install fail here, and converge
hits the same dirent at line 3178 and fails too, so the partition fences and
rebuilds forever. keep it non-fatal, like the purge path at
`iggy_partition.rs:6410`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2008,83 +2120,122 @@ 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);
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 offset <=
state.persisted_high_water => {
+ state.persisted_high_water
}
- None => {
- persist_offset_max(&path, offset,
self.consumer_offset_enforce_fsync)
- .await?
+ (Some(path), Some(state)) => {
+ let value = state.committed_offset.max(offset);
+ persist_offset(path, value,
self.consumer_offset_enforce_fsync).await?;
+ value
+ }
+ (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,
+ );
+ if path.is_some() && self.consumer_offset_enforce_fsync {
+ self.mark_consumer_offset_dir_dirty(pending.kind);
+ }
+ capacity.clear_stranded(pending.consumer_id);
+ if pending.kind == ConsumerKind::ConsumerGroup &&
tracked.is_none() {
+ self.mark_consumer_group_offsets_need_reconcile();
+ }
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?;
+ }
+ let created = self.durable_consumer_offsets.record_explicit(
+ pending.kind,
+ pending.consumer_id,
+ offset,
+ offset,
+ );
+ if path.is_some() && self.consumer_offset_enforce_fsync {
+ self.mark_consumer_offset_dir_dirty(pending.kind);
+ }
+ capacity.clear_stranded(pending.consumer_id);
+ if pending.kind == ConsumerKind::ConsumerGroup && created {
+ self.mark_consumer_group_offsets_need_reconcile();
+ }
Ok(())
}
PendingConsumerOffsetMutation::Delete => {
- delete_persisted_offset(&path).await?;
- self.persisted_offsets.borrow_mut().remove(&key);
+ if let Some(path) = path.as_deref() {
+ if delete_persisted_offset(path).await? {
Review Comment:
warning: a committed delete whose unlink fails takes the process down via
`FatalCommit`, and the file survives so boot undoes the delete anyway. strand
it and return `Ok`, and suppress the map removal too or `snapshot_offset_kind`
marks the partition inconsistent forever.
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -1324,7 +1324,13 @@ impl TcpClient {
// never reach the rest of the roster.
(next, true)
} else {
- return Err(IggyError::TransientNotAccepted);
+ // A one-node roster has nowhere else to walk while a
+ // freshly committed partition is still materialising.
+ // The server explicitly did not admit this request, so
+ // keep retrying the current endpoint within the
existing
+ // overall deadline rather than surfacing a transient
+ // solely because the roster contains no alternative.
+ (current, false)
};
loop {
Review Comment:
warning: this keeps `routing_guard` held to the 30s deadline instead of
dropping it when the walk exhausts, so other transient requests block at line
1284 and reconnect stalls on the same lock. drop the guard before re-issuing.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2008,83 +2120,122 @@ 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);
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 offset <=
state.persisted_high_water => {
+ state.persisted_high_water
}
- None => {
- persist_offset_max(&path, offset,
self.consumer_offset_enforce_fsync)
- .await?
+ (Some(path), Some(state)) => {
+ let value = state.committed_offset.max(offset);
+ persist_offset(path, value,
self.consumer_offset_enforce_fsync).await?;
+ value
+ }
+ (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,
+ );
+ if path.is_some() && self.consumer_offset_enforce_fsync {
Review Comment:
warning: `path.is_some()` does not mean a file was written - line 2152
writes nothing, and line 2161 skips its write when the file already covers the
value. key the flag on what persist actually returned.
##########
core/partitions/src/consumer_offset_capacity.rs:
##########
@@ -0,0 +1,658 @@
+// 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, Ref, RefCell};
+use std::collections::hash_map::Entry;
+use std::collections::{HashMap, HashSet};
+use std::rc::Rc;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DurableOffsetState {
+ pub(crate) committed_offset: u64,
+ pub(crate) persisted_high_water: u64,
+}
+
+#[derive(Debug, Default)]
+pub struct DurableConsumerOffsets {
+ consumers: RefCell<HashMap<u32, DurableOffsetState>>,
+ groups: RefCell<HashMap<u32, DurableOffsetState>>,
+ membership_epoch: Cell<u64>,
+}
+
+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 >=
offset
+ })
+ }
+
+ pub(crate) fn record_explicit(
+ &self,
+ kind: ConsumerKind,
+ id: u32,
+ committed_offset: u64,
+ persisted_high_water: u64,
+ ) -> bool {
+ let created = self
+ .entries(kind)
+ .borrow_mut()
+ .insert(
+ id,
+ DurableOffsetState {
+ committed_offset,
+ persisted_high_water,
+ },
+ )
+ .is_none();
+ if created {
+ self.bump_membership_epoch();
+ }
+ created
+ }
+
+ 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();
+ match entries.entry(id) {
+ Entry::Occupied(mut entry) => {
+ let state = entry.get_mut();
+ state.committed_offset =
state.committed_offset.max(committed_offset);
+ state.persisted_high_water =
state.persisted_high_water.max(persisted_high_water);
+ }
+ Entry::Vacant(entry) => {
+ entry.insert(DurableOffsetState {
+ committed_offset,
+ persisted_high_water,
+ });
+ drop(entries);
+ self.bump_membership_epoch();
+ }
+ }
+ }
+
+ pub(crate) fn remove(&self, kind: ConsumerKind, id: u32) -> bool {
+ let removed = self.entries(kind).borrow_mut().remove(&id).is_some();
+ if removed {
+ self.bump_membership_epoch();
+ }
+ removed
+ }
+
+ pub(crate) fn clear(&self) {
+ self.consumers.borrow_mut().clear();
+ self.groups.borrow_mut().clear();
+ self.bump_membership_epoch();
+ }
+
+ #[cfg(any(test, feature = "simulator"))]
+ pub(crate) fn committed_entries(&self, kind: ConsumerKind) -> Vec<(u32,
u64)> {
+ self.entries(kind)
+ .borrow()
+ .iter()
+ .map(|(id, state)| (*id, state.committed_offset))
+ .collect()
+ }
+
+ pub(crate) fn snapshot_entries(
+ &self,
+ kind: ConsumerKind,
+ ) -> Ref<'_, HashMap<u32, DurableOffsetState>> {
+ self.entries(kind).borrow()
+ }
+
+ const fn entries(&self, kind: ConsumerKind) -> &RefCell<HashMap<u32,
DurableOffsetState>> {
+ match kind {
+ ConsumerKind::Consumer => &self.consumers,
+ ConsumerKind::ConsumerGroup => &self.groups,
+ }
+ }
+
+ fn bump_membership_epoch(&self) {
+ self.membership_epoch
+ .set(self.membership_epoch.get().wrapping_add(1));
+ }
+}
+
+#[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, Arc<ProvisionalToken>>>,
+ stranded: RefCell<HashSet<u32>>,
+ uncertain: Cell<bool>,
+ durable_warned: Cell<bool>,
+ map_warned: Cell<bool>,
+ reclaim_epoch: Arc<AtomicU64>,
+ last_reclaim: Cell<Option<(u64, u64)>>,
+}
+
+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),
+ reclaim_epoch: Arc::new(AtomicU64::new(0)),
+ last_reclaim: Cell::new(None),
+ }
+ }
+
+ 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(
Review Comment:
warning: `occupied()` walks pending, provisional and stranded on every
new-key admission, so ramping n consumers is quadratic on the shard thread.
when the four counts sum below the limit, admit without walking - but keep the
`uncertain` gate.
##########
core/partitions/src/offset_storage.rs:
##########
@@ -109,6 +112,54 @@ pub fn decode_offset_record(bytes: &[u8]) -> OffsetRecord {
/// # Errors
/// [`IggyError`] when the directory, file, or write cannot be created or
completed.
pub async fn persist_offset(path: &str, offset: u64, enforce_fsync: bool) ->
Result<(), IggyError> {
+ replace_file(path, encode_offset_record(offset), enforce_fsync,
false).await
+}
+
+pub(crate) async fn stage_offset_replacement(path: &str, offset: u64) ->
Result<(), IggyError> {
Review Comment:
nit: hardcoding `enforce_fsync = true` here is right, since install unlinks
then renames and non-durable staging would lose every cursor on a crash. say so
in a comment - it reads like the knob was missed.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2261,21 +2418,44 @@ where
);
if let Err(error) = self.persist_consumer_offset_commit(pending).await
{
+ if offset.is_some() {
+ self.release_consumer_offset_reservation(kind, consumer_id);
+ }
emit_partition_diag(
tracing::Level::WARN,
&PartitionDiagEvent::new(self.diag_ctx(), "no_ack offset
persist failed")
.with_operation(request_header.operation)
.with_error(error.to_string()),
);
+ Self::send_partition_deny_or_log(
+ &self.consensus,
+ &request_header,
+ error.as_code(),
+ "no_ack offset failure reply send failed",
+ waiter,
+ )
+ .await;
return;
}
- if let Err(error) = self.apply_consumer_offset_commit(pending) {
+ self.apply_consumer_offset_commit(pending);
Review Comment:
warning: apply runs before the dir sync, so a sync failure denies the client
after the offset is already visible. a delete then retries into 3021 for a
delete that took effect - document the contract or roll back.
##########
core/server/src/offset_recovery.rs:
##########
@@ -167,28 +201,43 @@ pub fn load_consumer_group_offsets(
consumer_group_offsets.push((consumer_group_id, consumer_offset));
}
- Ok(consumer_group_offsets)
+ Ok(RecoveredOffsets {
+ entries: consumer_group_offsets,
+ stranded_ids: stranded,
+ })
}
-fn read_offset_file(path: &str, offset_kind: &'static str) ->
Option<AtomicU64> {
+/// A crashed atomic replacement leaves its sibling behind. The rename never
+/// landed, so the sibling carries nothing the numeric file lacks.
+fn remove_stale_replacement(path: &std::path::Path, name: &str) {
Review Comment:
nit: this unlinks with no parent fsync and only warns, while
`remove_invalid_offset_file` below fsyncs and downgrades to stranded on
failure. fine either way, but say why the replacement sibling needs no durable
removal.
##########
core/shard/src/lib.rs:
##########
@@ -8310,18 +8354,16 @@ where
if !commit_lag && head <= commit_to_op {
return false;
}
- let missing_suffix = partition_missing_suffix(partition);
- if !commit_lag && !missing_suffix {
+ let missing_suffix = partition_missing_suffix_through(partition);
+ if !commit_lag && missing_suffix.is_none() {
return false;
}
let nonce = iggy_common::random_id::get_uuid();
let from_op = consensus.commit_min() + 1;
- // The widening is for the replica that is LEVEL with the commit
- // frontier and short of bodies above it. Widening while a commit lag
- // stands would ask for `(commit_min, head]` -- the whole committed
- // prefix this replica already holds, refetched -- and the suffix is
- // reached anyway once the lag closes, on the arm after it.
- let fetch_to_op = if commit_lag { commit_to_op } else { head };
+ // Include the adopted suffix in the same repair as the committed
+ // prefix. Otherwise a newer live prepare can advance the sequencer
+ // while an older adopted body is still missing.
+ let fetch_to_op = missing_suffix.unwrap_or(commit_to_op);
Review Comment:
nit: this widens `fetch_to_op` to the adopted head even under commit lag,
reversing the comment you deleted. it is a real liveness fix and strictly
additive, but nothing asserts `fetch_to_op`, and line 9751 still links the
removed `partition_missing_suffix`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2341,7 +2610,21 @@ where
}
};
- if found {
+ if self.consensus.replica_count() > 1 {
Review Comment:
nit: a replicated delete for a live-map key that is not durable answers 3021
while `GetConsumerOffset` still returns its value. the client sees a key it
cannot delete.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2587,7 +3065,7 @@ where
offset: u64,
) -> Result<(), IggyError> {
let pending =
PendingConsumerOffsetCommit::try_from_polling_consumer(consumer, offset)?;
- self.apply_consumer_offset_commit(pending)?;
+ self.apply_consumer_offset_commit(pending);
Review Comment:
nit: this applies without persisting, so it arms neither the reconcile flag
nor the shard epoch, and skips `persist_offset` so the commit dies on restart.
no production caller - delete it or route it through the staged path.
##########
core/server/src/consumer_group.rs:
##########
@@ -226,16 +225,21 @@ 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)?;
- let Some(group_id) =
- resolve_group_offset_id(shard, &wire.consumer,
(&wire.stream_id, &wire.topic_id))
- else {
+ if wire.consumer.kind != KIND_CONSUMER_GROUP {
return Ok(request);
- };
+ }
+ let group_id = resolve_offset_group_id(
Review Comment:
nit: a store or delete against a missing group now returns 5000/5003 instead
of `ResourceNotFound`. go clients matching `ErrResourceNotFound` reroute
silently - name both codes in the PR body.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2341,7 +2610,21 @@ where
}
};
Review Comment:
nit: this branch is the replicated default and it throws away `found`, which
cost two pins and two `contains_key` calls just above. move that computation
under this early return.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -1551,6 +1635,21 @@ fn final_paths(partition_dir: &str, start_offset: u64)
-> (String, String) {
/// depth against how long one partition monopolises it; matches the tick's
/// superblock pre-pass.
const OFFSET_PERSIST_CONCURRENCY: usize = 16;
+const OFFSET_IO_ATTEMPTS: usize = 3;
+
+async fn retry_offset_mutation<T, E, F: Future<Output = Result<T, E>>>(
Review Comment:
nit: three attempts with only a reactor yield between them finish in
microseconds against a hard ENOSPC or EIO, and the caller only sees the last
error. either back off properly or do not retry this class.
##########
core/partitions/src/consumer_offset_capacity.rs:
##########
@@ -0,0 +1,658 @@
+// 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, Ref, RefCell};
+use std::collections::hash_map::Entry;
+use std::collections::{HashMap, HashSet};
+use std::rc::Rc;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DurableOffsetState {
+ pub(crate) committed_offset: u64,
+ pub(crate) persisted_high_water: u64,
+}
+
+#[derive(Debug, Default)]
+pub struct DurableConsumerOffsets {
+ consumers: RefCell<HashMap<u32, DurableOffsetState>>,
+ groups: RefCell<HashMap<u32, DurableOffsetState>>,
+ membership_epoch: Cell<u64>,
+}
+
+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 >=
offset
+ })
+ }
+
+ pub(crate) fn record_explicit(
+ &self,
+ kind: ConsumerKind,
+ id: u32,
+ committed_offset: u64,
+ persisted_high_water: u64,
+ ) -> bool {
+ let created = self
+ .entries(kind)
+ .borrow_mut()
+ .insert(
+ id,
+ DurableOffsetState {
+ committed_offset,
+ persisted_high_water,
+ },
+ )
+ .is_none();
+ if created {
+ self.bump_membership_epoch();
+ }
+ created
+ }
+
+ 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();
+ match entries.entry(id) {
+ Entry::Occupied(mut entry) => {
+ let state = entry.get_mut();
+ state.committed_offset =
state.committed_offset.max(committed_offset);
+ state.persisted_high_water =
state.persisted_high_water.max(persisted_high_water);
+ }
+ Entry::Vacant(entry) => {
+ entry.insert(DurableOffsetState {
+ committed_offset,
+ persisted_high_water,
+ });
+ drop(entries);
+ self.bump_membership_epoch();
+ }
+ }
+ }
+
+ pub(crate) fn remove(&self, kind: ConsumerKind, id: u32) -> bool {
+ let removed = self.entries(kind).borrow_mut().remove(&id).is_some();
+ if removed {
+ self.bump_membership_epoch();
+ }
+ removed
+ }
+
+ pub(crate) fn clear(&self) {
+ self.consumers.borrow_mut().clear();
+ self.groups.borrow_mut().clear();
+ self.bump_membership_epoch();
+ }
+
+ #[cfg(any(test, feature = "simulator"))]
+ pub(crate) fn committed_entries(&self, kind: ConsumerKind) -> Vec<(u32,
u64)> {
+ self.entries(kind)
+ .borrow()
+ .iter()
+ .map(|(id, state)| (*id, state.committed_offset))
+ .collect()
+ }
+
+ pub(crate) fn snapshot_entries(
+ &self,
+ kind: ConsumerKind,
+ ) -> Ref<'_, HashMap<u32, DurableOffsetState>> {
+ self.entries(kind).borrow()
+ }
+
+ const fn entries(&self, kind: ConsumerKind) -> &RefCell<HashMap<u32,
DurableOffsetState>> {
+ match kind {
+ ConsumerKind::Consumer => &self.consumers,
+ ConsumerKind::ConsumerGroup => &self.groups,
+ }
+ }
+
+ fn bump_membership_epoch(&self) {
+ self.membership_epoch
+ .set(self.membership_epoch.get().wrapping_add(1));
+ }
+}
+
+#[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, Arc<ProvisionalToken>>>,
+ stranded: RefCell<HashSet<u32>>,
+ uncertain: Cell<bool>,
+ durable_warned: Cell<bool>,
Review Comment:
simplification: `rearm_map_if_below_limit` is redundant once
`admit_local_map_key` clears the latch on success - its only `Ok` path is
already `map_len < limit`. drop the method and its eight call sites.
##########
core/server/src/offset_recovery.rs:
##########
@@ -27,19 +27,31 @@
//! here as unchecksummed.
use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset, IggyError};
-use partitions::offset_storage::{OffsetRecord, decode_offset_record};
+use partitions::offset_storage::{OFFSET_REPLACEMENT_SUFFIX, OffsetRecord,
decode_offset_record};
use std::sync::atomic::AtomicU64;
use tracing::{error, trace, warn};
const COMPONENT: &str = "STREAMING_PARTITIONS";
-pub fn load_consumer_offsets(path: &str) -> Result<Vec<ConsumerOffset>,
IggyError> {
+pub struct RecoveredOffsets<T> {
+ pub entries: Vec<T>,
+ pub stranded_ids: Vec<u32>,
+}
+
+enum OffsetFileLoad {
+ Loaded(AtomicU64),
+ Removed,
+ Stranded,
+}
+
+pub fn load_consumer_offsets(path: &str) ->
Result<RecoveredOffsets<ConsumerOffset>, IggyError> {
Review Comment:
simplification: this and `load_consumer_group_offsets` are the same 60-line
walk with a different element. one private helper taking a constructor closure
saves about 55 lines - note the consumer side sorts and the group side does not.
##########
core/sdk/src/quic/quic_client.rs:
##########
@@ -218,7 +218,10 @@ impl BinaryTransport for QuicClient {
} else if let Some(next) =
roster_walk.as_mut().and_then(RosterWalk::next) {
(next, true)
} else {
Review Comment:
warning: same as the tcp client - this fires on any exhausted roster walk,
not just a one-node roster, so a full-cluster refusal replays one endpoint
instead of surfacing `TransientNotAccepted`. gate on roster size.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2355,14 +2638,177 @@ 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) {
+ self.resynchronize_consumer_offset_reservations_inner(false);
+ }
+
+ /// Retry incomplete accounting at most once per shard tick, after
progress.
+ pub fn retry_consumer_offset_reservations(&mut self) {
+ if self.consumer_offset_capacity.is_uncertain()
+ || self.consumer_group_offset_capacity.is_uncertain()
+ {
+ self.resynchronize_consumer_offset_reservations_inner(true);
+ }
+ }
+
+ #[allow(clippy::too_many_lines)]
+ fn resynchronize_consumer_offset_reservations_inner(&mut self, from_tick:
bool) {
let current_view = self.consensus.view();
+ let scan_state = (
+ self.consensus.commit_min(),
+ self.consensus.commit_max(),
+ self.consensus.sequencer().current_sequence(),
+ self.log.journal().inner.last_op(),
+ );
+ let uncertain = self.consumer_offset_capacity.is_uncertain()
+ || self.consumer_group_offset_capacity.is_uncertain();
if current_view == self.observed_view {
- return;
+ if uncertain && !from_tick {
+ return;
+ }
+ let retry_requested = self.offset_reservations_need_resync.get()
+ || (uncertain && self.offset_reservations_scan_state !=
Some(scan_state));
+ if !retry_requested {
+ 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 commit_max = self.consensus.commit_max();
+ let to_op = self
+ .consensus
+ .sequencer()
+ .current_sequence()
+ .min(self.log.journal().inner.last_op().unwrap_or(commit_max));
+ // Committed offset prepares still need local apply, even if a message
+ // flush already evicted their journal bytes. Never drop their staging.
+ let mut rebuilt: HashMap<_, _> = self
+ .pending_consumer_offset_commits
+ .iter()
+ .filter(|(op, _)| **op >= from_op && **op <= commit_max)
+ .map(|(op, pending)| (*op, *pending))
+ .collect();
+ let headers =
self.log.journal().inner.repair_headers_in(from_op..=to_op);
+ let uncommitted_from = from_op.max(commit_max.saturating_add(1));
+ let expected = to_op
+ .checked_sub(uncommitted_from)
+ .map_or(0, |span| span.saturating_add(1));
+ let mut decode_failed =
+ headers.keys().filter(|op| **op >= uncommitted_from).count() 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.mark_consumer_group_offsets_need_reconcile();
Review Comment:
warning: bumping the shard-wide epoch unconditionally defeats the O(1)
reconciler skip added at `partition_reconciler.rs:495`, for every partition on
the shard, including ones holding no group offsets. bump it inside the
view-change check at line 2683.
##########
core/server/src/responses.rs:
##########
@@ -295,10 +293,46 @@ where
false,
)
.map(|_| ())
- .ok_or(IggyError::ConsumerGroupPartitionNotOwned(
- client_id as u32,
- partition_id,
- ))
+ .ok_or_else(|| {
Review Comment:
simplification: the nested `ok_or_else` resolves the group twice on the
error path. flatten it with a `let ... else` - do not reorder to resolve first,
that adds a walk to the success path.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -4470,6 +5092,56 @@ where
self.drain_request_queue_into_prepares(drained_count).await;
}
+ async fn flush_consumer_offset_directories(&self) -> Result<(), IggyError>
{
Review Comment:
warning: this returns on the first failing index and never tries the second,
so a groups dir fault leaves consumer dirents unproven. attempt both, clear
each on its own success, report after the loop.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2355,14 +2638,177 @@ 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) {
+ self.resynchronize_consumer_offset_reservations_inner(false);
+ }
+
+ /// Retry incomplete accounting at most once per shard tick, after
progress.
+ pub fn retry_consumer_offset_reservations(&mut self) {
+ if self.consumer_offset_capacity.is_uncertain()
+ || self.consumer_group_offset_capacity.is_uncertain()
+ {
+ self.resynchronize_consumer_offset_reservations_inner(true);
+ }
+ }
+
+ #[allow(clippy::too_many_lines)]
+ fn resynchronize_consumer_offset_reservations_inner(&mut self, from_tick:
bool) {
let current_view = self.consensus.view();
+ let scan_state = (
+ self.consensus.commit_min(),
+ self.consensus.commit_max(),
+ self.consensus.sequencer().current_sequence(),
+ self.log.journal().inner.last_op(),
+ );
+ let uncertain = self.consumer_offset_capacity.is_uncertain()
+ || self.consumer_group_offset_capacity.is_uncertain();
if current_view == self.observed_view {
- return;
+ if uncertain && !from_tick {
+ return;
+ }
+ let retry_requested = self.offset_reservations_need_resync.get()
+ || (uncertain && self.offset_reservations_scan_state !=
Some(scan_state));
+ if !retry_requested {
+ 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 commit_max = self.consensus.commit_max();
+ let to_op = self
+ .consensus
+ .sequencer()
+ .current_sequence()
+ .min(self.log.journal().inner.last_op().unwrap_or(commit_max));
+ // Committed offset prepares still need local apply, even if a message
+ // flush already evicted their journal bytes. Never drop their staging.
+ let mut rebuilt: HashMap<_, _> = self
+ .pending_consumer_offset_commits
+ .iter()
+ .filter(|(op, _)| **op >= from_op && **op <= commit_max)
+ .map(|(op, pending)| (*op, *pending))
+ .collect();
+ let headers =
self.log.journal().inner.repair_headers_in(from_op..=to_op);
+ let uncommitted_from = from_op.max(commit_max.saturating_add(1));
+ let expected = to_op
+ .checked_sub(uncommitted_from)
+ .map_or(0, |span| span.saturating_add(1));
+ let mut decode_failed =
+ headers.keys().filter(|op| **op >= uncommitted_from).count() 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.mark_consumer_group_offsets_need_reconcile();
+ self.offset_reservations_scan_state = Some(scan_state);
+ // The shard tick retries uncertainty after journal or frontier
progress.
+ self.offset_reservations_need_resync.set(false);
+ }
+
+ fn reclaim_phantom_offsets(&self, kind: ConsumerKind, map_count: usize) {
+ let capacity = self.consumer_offset_capacity_for(kind);
+ if self.durable_consumer_offsets.count(kind) >= capacity.limit()
+ || !capacity.should_reclaim(&self.durable_consumer_offsets)
+ {
+ return;
+ }
+ let mut remaining =
map_count.saturating_sub(capacity.limit()).saturating_add(1);
+ match kind {
Review Comment:
simplification: the two arms are the same loop over a different map. a
generic helper needs only `K: Hash + Eq` - roughly line-neutral, but it gives
one `break`/`remaining` protocol and one place for the eviction log.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2669,15 +2894,21 @@ where
// and a purged origin offering `next_offset = 0` drops every
// incoming entry, so a map-only sweep leaves the old table for
boot
// to resurrect.
- paths.extend(strayed_offset_files(
- self.consumer_offsets_path.as_deref(),
- &offsets_wire.consumers,
- ));
+ paths.extend(
+ strayed_offset_files(
Review Comment:
warning: this excludes `offsets_wire` ids, but `clamp` drops every entry
when `next_offset` is 0, so nothing renames them back and an old offset file
survives unowned. boot then resurrects it - build the set from
`planned_offsets` ids instead.
##########
core/partitions/src/offset_storage.rs:
##########
@@ -109,6 +112,54 @@ pub fn decode_offset_record(bytes: &[u8]) -> OffsetRecord {
/// # Errors
/// [`IggyError`] when the directory, file, or write cannot be created or
completed.
pub async fn persist_offset(path: &str, offset: u64, enforce_fsync: bool) ->
Result<(), IggyError> {
Review Comment:
warning: every durable offset commit now burns a fresh inode and a
`renameat` instead of an in-place truncate, plus a batched dir fsync when
`enforce_fsync` is on. that's the auto-commit hot path - note the trade-off
here and measure it.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2355,14 +2638,177 @@ 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) {
+ self.resynchronize_consumer_offset_reservations_inner(false);
+ }
+
+ /// Retry incomplete accounting at most once per shard tick, after
progress.
+ pub fn retry_consumer_offset_reservations(&mut self) {
+ if self.consumer_offset_capacity.is_uncertain()
+ || self.consumer_group_offset_capacity.is_uncertain()
+ {
+ self.resynchronize_consumer_offset_reservations_inner(true);
+ }
+ }
+
+ #[allow(clippy::too_many_lines)]
+ fn resynchronize_consumer_offset_reservations_inner(&mut self, from_tick:
bool) {
let current_view = self.consensus.view();
+ let scan_state = (
+ self.consensus.commit_min(),
+ self.consensus.commit_max(),
+ self.consensus.sequencer().current_sequence(),
+ self.log.journal().inner.last_op(),
+ );
+ let uncertain = self.consumer_offset_capacity.is_uncertain()
+ || self.consumer_group_offset_capacity.is_uncertain();
if current_view == self.observed_view {
- return;
+ if uncertain && !from_tick {
+ return;
+ }
+ let retry_requested = self.offset_reservations_need_resync.get()
+ || (uncertain && self.offset_reservations_scan_state !=
Some(scan_state));
+ if !retry_requested {
+ 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 commit_max = self.consensus.commit_max();
+ let to_op = self
+ .consensus
+ .sequencer()
+ .current_sequence()
+ .min(self.log.journal().inner.last_op().unwrap_or(commit_max));
+ // Committed offset prepares still need local apply, even if a message
+ // flush already evicted their journal bytes. Never drop their staging.
+ let mut rebuilt: HashMap<_, _> = self
+ .pending_consumer_offset_commits
+ .iter()
+ .filter(|(op, _)| **op >= from_op && **op <= commit_max)
+ .map(|(op, pending)| (*op, *pending))
+ .collect();
+ let headers =
self.log.journal().inner.repair_headers_in(from_op..=to_op);
+ let uncommitted_from = from_op.max(commit_max.saturating_add(1));
+ let expected = to_op
+ .checked_sub(uncommitted_from)
+ .map_or(0, |span| span.saturating_add(1));
+ let mut decode_failed =
+ headers.keys().filter(|op| **op >= uncommitted_from).count() 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.mark_consumer_group_offsets_need_reconcile();
+ self.offset_reservations_scan_state = Some(scan_state);
+ // The shard tick retries uncertainty after journal or frontier
progress.
+ self.offset_reservations_need_resync.set(false);
+ }
+
+ fn reclaim_phantom_offsets(&self, kind: ConsumerKind, map_count: usize) {
Review Comment:
warning: eviction drops live cursors silently and in hash order, so a hot
consumer can lose its cursor and re-read the partition from 0 while a colder
one survives. add a `debug!` with the kind and consumer id.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2649,18 +2872,20 @@ where
let next_offset = offsets_wire
.next_offset
.max(installed_end.map_or(0, |end| end + 1));
- let mut offsets_written = true;
// A key that fails the u32 narrowing would strand its old offset
// file's delete, which boot can then resurrect: unreachable while
// keys are minted from u32 wire ids, so assert it.
- let old_consumer_paths: Vec<String> = {
+ let old_consumer_paths: Vec<(ConsumerKind, u32, String)> = {
Review Comment:
nit: ids in both the map and the incoming table are unlinked here then
re-created by rename, so a crash between the two loses the cursor. filter on
ids that have a planned write, not on `offsets_wire` ids.
##########
core/partitions/src/consumer_offset_capacity.rs:
##########
@@ -0,0 +1,658 @@
+// 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, Ref, RefCell};
+use std::collections::hash_map::Entry;
+use std::collections::{HashMap, HashSet};
+use std::rc::Rc;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DurableOffsetState {
+ pub(crate) committed_offset: u64,
+ pub(crate) persisted_high_water: u64,
+}
+
+#[derive(Debug, Default)]
+pub struct DurableConsumerOffsets {
+ consumers: RefCell<HashMap<u32, DurableOffsetState>>,
+ groups: RefCell<HashMap<u32, DurableOffsetState>>,
+ membership_epoch: Cell<u64>,
+}
+
+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 >=
offset
+ })
+ }
+
+ pub(crate) fn record_explicit(
+ &self,
+ kind: ConsumerKind,
+ id: u32,
+ committed_offset: u64,
+ persisted_high_water: u64,
+ ) -> bool {
+ let created = self
+ .entries(kind)
+ .borrow_mut()
+ .insert(
+ id,
+ DurableOffsetState {
+ committed_offset,
+ persisted_high_water,
+ },
+ )
+ .is_none();
+ if created {
+ self.bump_membership_epoch();
+ }
+ created
+ }
+
+ 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();
+ match entries.entry(id) {
+ Entry::Occupied(mut entry) => {
+ let state = entry.get_mut();
+ state.committed_offset =
state.committed_offset.max(committed_offset);
+ state.persisted_high_water =
state.persisted_high_water.max(persisted_high_water);
+ }
+ Entry::Vacant(entry) => {
+ entry.insert(DurableOffsetState {
+ committed_offset,
+ persisted_high_water,
+ });
+ drop(entries);
+ self.bump_membership_epoch();
+ }
+ }
+ }
+
+ pub(crate) fn remove(&self, kind: ConsumerKind, id: u32) -> bool {
+ let removed = self.entries(kind).borrow_mut().remove(&id).is_some();
+ if removed {
+ self.bump_membership_epoch();
+ }
+ removed
+ }
+
+ pub(crate) fn clear(&self) {
+ self.consumers.borrow_mut().clear();
+ self.groups.borrow_mut().clear();
+ self.bump_membership_epoch();
+ }
+
+ #[cfg(any(test, feature = "simulator"))]
+ pub(crate) fn committed_entries(&self, kind: ConsumerKind) -> Vec<(u32,
u64)> {
+ self.entries(kind)
+ .borrow()
+ .iter()
+ .map(|(id, state)| (*id, state.committed_offset))
+ .collect()
+ }
+
+ pub(crate) fn snapshot_entries(
Review Comment:
warning: `snapshot_entries` hands a live `Ref` out of the type and the
caller runs a closure while it is held, so any future closure touching
`durable_consumer_offsets` panics the pump on a double borrow. invert it to
`with_entries(kind, |entries| ..)`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2355,14 +2638,177 @@ 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) {
+ self.resynchronize_consumer_offset_reservations_inner(false);
+ }
+
+ /// Retry incomplete accounting at most once per shard tick, after
progress.
+ pub fn retry_consumer_offset_reservations(&mut self) {
+ if self.consumer_offset_capacity.is_uncertain()
+ || self.consumer_group_offset_capacity.is_uncertain()
+ {
+ self.resynchronize_consumer_offset_reservations_inner(true);
+ }
+ }
+
+ #[allow(clippy::too_many_lines)]
+ fn resynchronize_consumer_offset_reservations_inner(&mut self, from_tick:
bool) {
let current_view = self.consensus.view();
+ let scan_state = (
+ self.consensus.commit_min(),
+ self.consensus.commit_max(),
+ self.consensus.sequencer().current_sequence(),
+ self.log.journal().inner.last_op(),
+ );
+ let uncertain = self.consumer_offset_capacity.is_uncertain()
+ || self.consumer_group_offset_capacity.is_uncertain();
if current_view == self.observed_view {
- return;
+ if uncertain && !from_tick {
+ return;
+ }
+ let retry_requested = self.offset_reservations_need_resync.get()
+ || (uncertain && self.offset_reservations_scan_state !=
Some(scan_state));
+ if !retry_requested {
+ 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 commit_max = self.consensus.commit_max();
+ let to_op = self
+ .consensus
+ .sequencer()
+ .current_sequence()
+ .min(self.log.journal().inner.last_op().unwrap_or(commit_max));
+ // Committed offset prepares still need local apply, even if a message
+ // flush already evicted their journal bytes. Never drop their staging.
+ let mut rebuilt: HashMap<_, _> = self
+ .pending_consumer_offset_commits
+ .iter()
+ .filter(|(op, _)| **op >= from_op && **op <= commit_max)
+ .map(|(op, pending)| (*op, *pending))
+ .collect();
+ let headers =
self.log.journal().inner.repair_headers_in(from_op..=to_op);
Review Comment:
nit: while a partition is uncertain and advancing, every tick rebuilds the
whole staging window and deep-copies each offset op it finds. bounded by
journal residency, but worth an op budget with a resume cursor.
##########
core/partitions/src/offset_storage.rs:
##########
@@ -325,14 +364,15 @@ async fn read_offset_record(path: &str) ->
Result<Option<OffsetRecord>, IggyErro
/// Unlink a persisted consumer-offset file. A no-op if the file is absent.
Review Comment:
nit: the "returns whether a file was removed" line sits under `# Errors`, so
rustdoc renders success semantics as an error note. move it into the summary.
##########
core/server/src/offset_recovery.rs:
##########
@@ -63,11 +75,15 @@ pub fn load_consumer_offsets(path: &str) ->
Result<Vec<ConsumerOffset>, IggyErro
}
};
- if metadata.is_dir() {
+ if !metadata.is_file() {
continue;
}
let name = dir_entry.file_name().to_string_lossy().to_string();
+ if name.ends_with(OFFSET_REPLACEMENT_SUFFIX) {
Review Comment:
nit: this matches any `.tmp` in the offsets dir before parsing the id, so it
deletes unrelated files. require the stem to parse as `u32` first.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2355,14 +2638,177 @@ 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) {
+ self.resynchronize_consumer_offset_reservations_inner(false);
+ }
+
+ /// Retry incomplete accounting at most once per shard tick, after
progress.
+ pub fn retry_consumer_offset_reservations(&mut self) {
+ if self.consumer_offset_capacity.is_uncertain()
+ || self.consumer_group_offset_capacity.is_uncertain()
+ {
+ self.resynchronize_consumer_offset_reservations_inner(true);
+ }
+ }
+
+ #[allow(clippy::too_many_lines)]
+ fn resynchronize_consumer_offset_reservations_inner(&mut self, from_tick:
bool) {
let current_view = self.consensus.view();
+ let scan_state = (
+ self.consensus.commit_min(),
+ self.consensus.commit_max(),
+ self.consensus.sequencer().current_sequence(),
+ self.log.journal().inner.last_op(),
+ );
+ let uncertain = self.consumer_offset_capacity.is_uncertain()
+ || self.consumer_group_offset_capacity.is_uncertain();
if current_view == self.observed_view {
- return;
+ if uncertain && !from_tick {
+ return;
+ }
+ let retry_requested = self.offset_reservations_need_resync.get()
+ || (uncertain && self.offset_reservations_scan_state !=
Some(scan_state));
+ if !retry_requested {
+ 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 commit_max = self.consensus.commit_max();
+ let to_op = self
+ .consensus
+ .sequencer()
+ .current_sequence()
+ .min(self.log.journal().inner.last_op().unwrap_or(commit_max));
+ // Committed offset prepares still need local apply, even if a message
+ // flush already evicted their journal bytes. Never drop their staging.
+ let mut rebuilt: HashMap<_, _> = self
+ .pending_consumer_offset_commits
+ .iter()
+ .filter(|(op, _)| **op >= from_op && **op <= commit_max)
+ .map(|(op, pending)| (*op, *pending))
+ .collect();
+ let headers =
self.log.journal().inner.repair_headers_in(from_op..=to_op);
+ let uncommitted_from = from_op.max(commit_max.saturating_add(1));
+ let expected = to_op
+ .checked_sub(uncommitted_from)
+ .map_or(0, |span| span.saturating_add(1));
+ let mut decode_failed =
+ headers.keys().filter(|op| **op >= uncommitted_from).count() 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.mark_consumer_group_offsets_need_reconcile();
+ self.offset_reservations_scan_state = Some(scan_state);
+ // The shard tick retries uncertainty after journal or frontier
progress.
+ self.offset_reservations_need_resync.set(false);
+ }
+
+ fn reclaim_phantom_offsets(&self, kind: ConsumerKind, map_count: usize) {
+ let capacity = self.consumer_offset_capacity_for(kind);
+ if self.durable_consumer_offsets.count(kind) >= capacity.limit()
+ || !capacity.should_reclaim(&self.durable_consumer_offsets)
+ {
+ return;
+ }
+ let mut remaining =
map_count.saturating_sub(capacity.limit()).saturating_add(1);
+ match kind {
+ ConsumerKind::Consumer => {
+ let map = self.consumer_offsets.pin();
+ for (key, _) in &map {
+ if let Ok(id) = u32::try_from(*key)
+ && !capacity.holds(id, &self.durable_consumer_offsets)
+ {
+ map.remove(key);
+ capacity.forget_inactive_provisional(id);
+ remaining -= 1;
+ if remaining == 0 {
+ break;
+ }
+ }
+ }
+ }
+ ConsumerKind::ConsumerGroup => {
+ let map = self.consumer_group_offsets.pin();
+ for (key, _) in &map {
+ if let Ok(id) = u32::try_from(key.0)
+ && !capacity.holds(id, &self.durable_consumer_offsets)
+ {
+ map.remove(key);
+ capacity.forget_inactive_provisional(id);
+ remaining -= 1;
+ if remaining == 0 {
+ break;
+ }
+ }
+ }
+ }
+ }
+
capacity.rearm_map_if_below_limit(self.consumer_offset_map_count(kind));
Review Comment:
nit: this calls papaya `len()` even when the loop evicted nothing, and it
sums a per-CPU counter. `map_count` was read at line 2886 and `remaining`
tracks the rest, so subtract instead.
##########
core/server/src/offset_recovery.rs:
##########
@@ -167,28 +201,43 @@ pub fn load_consumer_group_offsets(
consumer_group_offsets.push((consumer_group_id, consumer_offset));
}
- Ok(consumer_group_offsets)
+ Ok(RecoveredOffsets {
+ entries: consumer_group_offsets,
+ stranded_ids: stranded,
+ })
}
-fn read_offset_file(path: &str, offset_kind: &'static str) ->
Option<AtomicU64> {
+/// A crashed atomic replacement leaves its sibling behind. The rename never
+/// landed, so the sibling carries nothing the numeric file lacks.
+fn remove_stale_replacement(path: &std::path::Path, name: &str) {
+ match std::fs::remove_file(path) {
+ Ok(()) => trace!("Removed stale offset replacement file: '{name}'."),
+ Err(e) => warn!(
+ "{COMPONENT} (error: {e}) - could not remove stale offset
replacement \
+ file: '{name}', skipping."
+ ),
+ }
+}
+
+fn read_offset_file(path: &str, offset_kind: &'static str) -> OffsetFileLoad {
let bytes = match std::fs::read(path) {
Ok(bytes) => bytes,
Err(e) => {
warn!(
"{COMPONENT} (error: {e}) - failed to read offset file, \
path: {path}, skipping."
);
- return None;
+ return OffsetFileLoad::Stranded;
}
};
match decode_offset_record(&bytes) {
- OffsetRecord::Value { offset, .. } => Some(AtomicU64::new(offset)),
+ OffsetRecord::Value { offset, .. } =>
OffsetFileLoad::Loaded(AtomicU64::new(offset)),
OffsetRecord::Torn => {
warn!(
Review Comment:
nit: the warn still ends `skipping.` but this branch unlinks the file now.
say removed.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2686,130 +2917,96 @@ where
"consumer group offset key {} exceeds u32",
key.0
);
- narrowed
- .and_then(|id|
self.persisted_offset_path(ConsumerKind::ConsumerGroup, id))
+ narrowed.and_then(|id| {
+
self.persisted_offset_path(ConsumerKind::ConsumerGroup, id)
+ .map(|path| (ConsumerKind::ConsumerGroup, id,
path))
+ })
})
.collect();
guard.clear();
- paths.extend(strayed_offset_files(
- self.consumer_group_offsets_path.as_deref(),
- &offsets_wire.groups,
- ));
+ paths.extend(
+ strayed_offset_files(
+ self.consumer_group_offsets_path.as_deref(),
+ &offsets_wire.groups,
+ )
+ .into_iter()
+ .filter_map(|path| {
+ numeric_offset_id(&path).map(|id|
(ConsumerKind::ConsumerGroup, id, path))
+ }),
+ );
paths
};
- for path in old_consumer_paths.into_iter().chain(old_group_paths) {
- if let Err(error) = delete_persisted_offset(&path).await {
- // Not fatal, but not silent either: a stranded file is an id
- // absent from the NEW table (matching ids get overwritten at
- // the same path), and boot resurrects it. Sharpest after a
- // purged origin ships `next_offset = 0`, where the clamp drops
- // every incoming entry and the whole old table survives while
- // the install still reports success.
- tracing::warn!(
- target: "iggy.partitions.diag",
- plane = "partitions",
- namespace_raw = self.consensus().group(),
- path = %path,
- %error,
- "failed to unlink a superseded consumer-offset file during
install"
- );
+ let mut offset_dirs_changed = [false; 2];
+ for (kind, consumer_id, path) in
old_consumer_paths.into_iter().chain(old_group_paths) {
+ let removed = delete_persisted_offset(&path)
+ .await
+ .map_err(|source| PartitionInstallError::OffsetPersistence {
path, source })?;
+ if removed {
+ offset_dirs_changed[consumer_kind_index(kind)] = true;
}
+ self.consumer_offset_capacity_for(kind)
+ .clear_stranded(consumer_id);
}
- self.persisted_offsets.borrow_mut().clear();
+ self.durable_consumer_offsets.clear();
self.pending_consumer_offset_commits.clear();
+ self.queued_auto_commit_reservations.borrow_mut().clear();
+ self.consumer_offset_capacity
+ .rebuild(&self.durable_consumer_offsets, std::iter::empty());
+ self.consumer_group_offset_capacity
+ .rebuild(&self.durable_consumer_offsets, std::iter::empty());
self.last_polled_offsets.pin().clear();
- // `None` when the group's offset space is empty (`next_offset == 0`,
- // a purged origin): clamping every transferred offset to 0 would tell
- // each consumer it consumed offset 0 on a partition that never minted
- // one, so a `Next` poll skips the first message. Dropping the entries
- // is what "no offsets yet" means.
- let clamp = |offset: u64| next_offset.checked_sub(1).map(|last|
offset.min(last));
- if self.consumer_offsets_path.is_none() ||
self.consumer_group_offsets_path.is_none() {
- // Nothing to write the transferred table into: unreachable via
- // the server boot paths (they always configure storage), but if
- // it ever fires the table was dropped and the flag must say so.
- offsets_written = false;
- }
- // Both maps are populated first (no await, so nothing borrows across
- // one), then the files are written in capped batches. One await per
- // file put a rejoin carrying thousands of consumers on the pump for
- // thousands of sequential open + write + optional fsync round trips;
- // the tick's superblock pre-pass sets the precedent for the width.
- // The dedup slice is memory-only, so it installs here with the maps
- // rather than being written anywhere. No frontier fence is needed: the
- // install lifts `commit_min` to the offer's `commit_op`, so the commit
- // walk that follows starts strictly above everything this artifact
- // covers, and `record_commit` is idempotent besides.
+ // The replacement siblings were written and data-synced before any
+ // segment mutation. Finalize only their directory entries here, then
+ // publish the matching maps and durable membership.
self.dedup_mut()
.install_watermarks(offsets_wire.dedup.iter().copied());
- let mut planned: Vec<PlannedOffsetWrite> =
- Vec::with_capacity(offsets_wire.consumers.len() +
offsets_wire.groups.len());
- if let Some(dir) = self.consumer_offsets_path.clone() {
- for (id, offset) in &offsets_wire.consumers {
- let Some(value) = clamp(*offset) else {
- continue;
- };
- let entry = ConsumerOffset::default_for_consumer(*id, &dir);
- entry.offset.store(value, Ordering::Release);
- let path = entry.path.clone();
- self.consumer_offsets.pin().insert(*id as usize, entry);
- planned.push(PlannedOffsetWrite {
- kind: ConsumerKind::Consumer,
- id: *id,
- path,
- value,
- });
- }
- }
- if let Some(dir) = self.consumer_group_offsets_path.clone() {
- for (id, offset) in &offsets_wire.groups {
- let Some(value) = clamp(*offset) else {
- continue;
- };
- let group_id = ConsumerGroupId(*id as usize);
- let entry =
ConsumerOffset::default_for_consumer_group(group_id, &dir);
- entry.offset.store(value, Ordering::Release);
- let path = entry.path.clone();
- self.consumer_group_offsets.pin().insert(group_id, entry);
- planned.push(PlannedOffsetWrite {
- kind: ConsumerKind::ConsumerGroup,
- id: *id,
- path,
- value,
- });
- }
- }
- let enforce_fsync = self.consumer_offset_enforce_fsync;
- for batch in planned.chunks(OFFSET_PERSIST_CONCURRENCY) {
- let writes = batch.iter().map(|write| async move {
- let written = persist_offset(&write.path, write.value,
enforce_fsync)
- .await
- .is_ok();
- (written, write.kind, write.id, write.value)
- });
- for (written, kind, id, value) in
futures::future::join_all(writes).await {
- if written {
- self.persisted_offsets
- .borrow_mut()
- .insert((kind, id), value);
- } else {
- offsets_written = false;
+ for write in planned_offsets {
+ commit_offset_replacement(&write.path)
Review Comment:
nit: the post-swap delete and rename are unguarded while the pre-swap
staging retries, and any errno here escalates to the full wipe-and-re-pull.
worth a note - a yield-only retry would not help this class.
##########
core/common/src/traits/consumer_offset_client.rs:
##########
@@ -24,6 +24,7 @@ pub trait ConsumerOffsetClient {
/// Store the consumer offset for a specific consumer or consumer group
for the given stream and topic by unique IDs or names.
///
/// Authentication is required, and the permission to poll the messages.
+ /// A new key at the per-partition limit returns
[`IggyError::TooManyConsumerOffsets`] (3024).
Review Comment:
nit: the 3024 note only covers `store_consumer_offset`.
`delete_consumer_offset` types the same admission errors, and
`get_consumer_offset` is where "missing key is `Ok(None)`" belongs.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2959,6 +3156,36 @@ where
}
self.log.journal().inner.clear_all();
self.log.journal_mut().info = crate::log::JournalInfo::default();
+ self.consumer_offsets.pin().clear();
+ self.consumer_group_offsets.pin().clear();
+ self.last_polled_offsets.pin().clear();
+ self.durable_consumer_offsets.clear();
+ self.pending_consumer_offset_commits.clear();
+ self.queued_auto_commit_reservations.borrow_mut().clear();
+ for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] {
+ self.consumer_offset_capacity_for(kind)
+ .rebuild(&self.durable_consumer_offsets, std::iter::empty());
+ }
+ for (kind, dir) in [
+ (ConsumerKind::Consumer, self.consumer_offsets_path.as_ref()),
+ (
+ ConsumerKind::ConsumerGroup,
+ self.consumer_group_offsets_path.as_ref(),
+ ),
+ ] {
+ let Some(dir) = dir else { continue };
+ for path in strayed_offset_files(Some(dir), &[]) {
Review Comment:
nit: `strayed_offset_files` parses names as `u32`, so the converge sweep
skips every `<id>.tmp` sibling even though the comment says the directory
converges to an empty shape.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -395,13 +457,15 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>(
operation = ?header.operation,
"partition request with unresolved namespace; replying denied"
Review Comment:
nit: the gate moved from error identity to operation identity, so every
namespace-resolution error on these two ops now goes to the client verbatim
instead of as `ResourceNotFound`. the wire-visible set widened with no coverage.
##########
core/server/src/offset_recovery.rs:
##########
@@ -206,13 +255,55 @@ fn read_offset_file(path: &str, offset_kind: &'static
str) -> Option<AtomicU64>
(offset: {offset}, expected: {expected}, found: {found}), \
path: {path}, removing it and resuming this consumer from the
start."
);
- if let Err(e) = std::fs::remove_file(path) {
- error!(
- "{COMPONENT} (error: {e}) - could not remove the corrupt \
- {offset_kind} file, path: {path}; remove it manually."
- );
- }
- None
+ remove_invalid_offset_file(path, offset_kind)
}
}
}
+
+fn remove_invalid_offset_file(path: &str, offset_kind: &'static str) ->
OffsetFileLoad {
Review Comment:
nit: this does a blocking `remove_file` plus parent `sync_all` inside the
loader, and `load_partition` runs from the reconciler at runtime, so a torn
file costs a dir fsync on a serving shard thread.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2008,83 +2120,122 @@ 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);
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 offset <=
state.persisted_high_water => {
+ state.persisted_high_water
}
- None => {
- persist_offset_max(&path, offset,
self.consumer_offset_enforce_fsync)
- .await?
+ (Some(path), Some(state)) => {
+ let value = state.committed_offset.max(offset);
+ persist_offset(path, value,
self.consumer_offset_enforce_fsync).await?;
+ value
+ }
+ (Some(path), None) => {
Review Comment:
nit: `persist_offset_max` skips its write when the file already covers the
value, so this branch can write nothing. the dirty-bit fix above needs its
predicate from the return value, not the branch.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -1577,6 +1710,45 @@ where
B: MessageBus,
SB: SuperblockStore,
{
+ fn plan_transfer_offset_writes(
+ &self,
+ offsets_wire: &ConsumerOffsetsWire,
+ next_offset: u64,
+ ) -> Result<Vec<PlannedOffsetWrite>, PartitionInstallError> {
+ let consumer_dir = self.consumer_offsets_path.as_deref().ok_or_else(||
{
+ PartitionInstallError::OffsetPersistence {
+ path: "consumer offset directory".to_owned(),
+ source: iggy_common::IggyError::InvalidConfiguration,
+ }
+ })?;
+ let group_dir =
self.consumer_group_offsets_path.as_deref().ok_or_else(|| {
+ PartitionInstallError::OffsetPersistence {
+ path: "consumer group offset directory".to_owned(),
+ source: iggy_common::IggyError::InvalidConfiguration,
+ }
+ })?;
+ let clamp = |offset: u64| next_offset.checked_sub(1).map(|last|
offset.min(last));
+ let mut planned =
+ Vec::with_capacity(offsets_wire.consumers.len() +
offsets_wire.groups.len());
+ planned.extend(offsets_wire.consumers.iter().filter_map(|(id, offset)|
{
Review Comment:
simplification: the two `planned.extend(..)` blocks differ only in kind, dir
and source vec. one closure applied twice.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -1577,6 +1710,45 @@ where
B: MessageBus,
SB: SuperblockStore,
{
+ fn plan_transfer_offset_writes(
+ &self,
+ offsets_wire: &ConsumerOffsetsWire,
+ next_offset: u64,
+ ) -> Result<Vec<PlannedOffsetWrite>, PartitionInstallError> {
+ let consumer_dir = self.consumer_offsets_path.as_deref().ok_or_else(||
{
Review Comment:
simplification: this fakes a path string to report a missing directory
config. a `NoOffsetDir` variant like `NoPartitionDir` drops two sentinels and a
`Display` that lies.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2186,38 +2323,58 @@ where
.collect()
}
- /// Reclaim every stored consumer-group offset whose group id is no longer
- /// `is_live`, returning the owned persisted-file paths the caller must
unlink.
- ///
- /// Fully synchronous (no `.await`): the in-memory papaya remove happens
here,
- /// the disk unlink is deferred to the caller on owned `String` data so no
- /// borrow of `self` survives across the await. This is the only safe shape
- /// for the reconciler, which runs on a sibling task to the pump that may
- /// realloc the partitions vec during that await. The remove-then-unlink
- /// ordering matches the crash-safe GC invariant (monotonic, never-reused
- /// group ids mean a recreated group never reads a dead group's offset).
+ /// Snapshot dead group keys for deletion through the partition's VSR log.
+ /// A local unlink could free the primary's quota while backups retained
+ /// every older generation, so reclamation uses the same ordered delete as
+ /// an explicit consumer-offset request.
#[must_use]
- #[allow(clippy::cast_possible_truncation)]
- pub fn reclaim_dead_group_offsets(&self, is_live: impl Fn(u64) -> bool) ->
Vec<String> {
- let pinned = self.consumer_group_offsets.pin();
- let dead: Vec<u64> = pinned
- .keys()
- .map(|key| key.0 as u64)
- .filter(|group_id| !is_live(*group_id))
- .collect();
- let mut paths = Vec::with_capacity(dead.len());
- for group_id in dead {
- pinned.remove(&ConsumerGroupId(group_id as usize));
- self.persisted_offsets
- .borrow_mut()
- .remove(&(ConsumerKind::ConsumerGroup, group_id as u32));
- if let Some(path) =
- self.persisted_offset_path(ConsumerKind::ConsumerGroup,
group_id as u32)
- {
- paths.push(path);
+ pub fn dead_consumer_group_offset_ids(&self, is_live: impl Fn(u64) ->
bool) -> Vec<u32> {
+ if !self.consensus.is_primary() || !self.consensus.is_normal() {
+ return Vec::new();
+ }
+ let mut dead = Vec::new();
+ for key in self.consumer_group_offsets.pin().keys() {
+ let Ok(id) = u32::try_from(key.0) else {
+ continue;
+ };
+ if !is_live(u64::from(id)) {
+ dead.push(id);
}
}
- paths
+ // A stranded file already failed normal loading or unlink. Reissuing
+ // replicated deletes every reconciliation pass cannot make its
+ // filesystem writable and would create a permanent commit loop.
+ // A single-replica explicit deletion can retry after repair. On a
+ // replicated partition the file must be repaired or removed locally,
+ // because older peers do not recognize a delete for a map-missing key.
+ dead.sort_unstable();
+ dead.dedup();
+ self.consumer_group_offsets_need_reconcile
+ .set(!dead.is_empty());
+ dead
+ }
+
+ #[must_use]
Review Comment:
simplification: `consumer_group_offsets_reconcile_needed()` has no caller
outside this file's tests - `reconcile_consumer_group_offsets` walks every
namespace ungated, so only the shard epoch does work. drop the reader and the
`Cell`.
##########
core/server/src/partition_helpers.rs:
##########
@@ -268,15 +313,21 @@ fn load_partition_consumer_offsets(
stream_id: usize,
topic_id: usize,
partition_id: usize,
-) -> Result<Vec<iggy_common::ConsumerOffset>, ServerError> {
+) -> Result<RecoveredOffsets<iggy_common::ConsumerOffset>, ServerError> {
if !Path::new(path).exists() {
Review Comment:
simplification: four copies of `RecoveredOffsets { entries: Vec::new(),
stranded_ids: Vec::new() }`. a hand-written `Default` (no `T: Default` bound)
removes all of them.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2649,18 +2872,20 @@ where
let next_offset = offsets_wire
Review Comment:
simplification: `next_offset` is computed here and again at line 2486. pass
it in - `installed_end` has to stay, line 3027 reads it.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -4470,6 +5092,56 @@ where
self.drain_request_queue_into_prepares(drained_count).await;
}
+ async fn flush_consumer_offset_directories(&self) -> Result<(), IggyError>
{
+ for (index, dir) in [
+ self.consumer_offsets_path.as_deref(),
+ self.consumer_group_offsets_path.as_deref(),
+ ]
+ .into_iter()
+ .enumerate()
+ {
+ if !self.consumer_offset_dirs_dirty[index].get() {
+ continue;
+ }
+ if let Some(dir) = dir {
+ match crate::state_transfer::fsync_dir(dir).await {
+ Ok(()) => {}
+ Err(error) if error.kind() == std::io::ErrorKind::NotFound
=> {
+ // The directory disappeared after the final unlink.
+ // There is no remaining dirent whose durability needs
+ // proving.
+ }
+ Err(error) => {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ replica_id = self.consensus.replica(),
+ namespace_raw = self.namespace().inner(),
+ path = dir,
+ error_kind = ?error.kind(),
+ %error,
+ "consumer offset directory sync failed"
+ );
+ return Err(IggyError::CannotSyncFile);
+ }
+ }
+ #[cfg(test)]
+ self.offset_dir_sync_count
+ .set(self.offset_dir_sync_count.get() + 1);
+ }
+ self.consumer_offset_dirs_dirty[index].set(false);
+ }
+ Ok(())
+ }
+
+ fn mark_consumer_offset_dir_dirty(&self, kind: ConsumerKind) {
+ let index = match kind {
Review Comment:
simplification: this repeats `consumer_kind_index` from
`state_transfer.rs:1664`, and both index `[_; 2]` arrays under the same
convention. make it `pub(crate)` and call it - two copies will drift.
##########
core/partitions/src/consumer_offset_capacity.rs:
##########
@@ -0,0 +1,658 @@
+// 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, Ref, RefCell};
+use std::collections::hash_map::Entry;
+use std::collections::{HashMap, HashSet};
+use std::rc::Rc;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DurableOffsetState {
+ pub(crate) committed_offset: u64,
+ pub(crate) persisted_high_water: u64,
+}
+
+#[derive(Debug, Default)]
+pub struct DurableConsumerOffsets {
+ consumers: RefCell<HashMap<u32, DurableOffsetState>>,
+ groups: RefCell<HashMap<u32, DurableOffsetState>>,
+ membership_epoch: Cell<u64>,
+}
+
+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 >=
offset
+ })
+ }
+
+ pub(crate) fn record_explicit(
+ &self,
+ kind: ConsumerKind,
+ id: u32,
+ committed_offset: u64,
+ persisted_high_water: u64,
+ ) -> bool {
+ let created = self
+ .entries(kind)
+ .borrow_mut()
+ .insert(
+ id,
+ DurableOffsetState {
+ committed_offset,
+ persisted_high_water,
+ },
+ )
+ .is_none();
+ if created {
+ self.bump_membership_epoch();
+ }
+ created
+ }
+
+ 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();
+ match entries.entry(id) {
+ Entry::Occupied(mut entry) => {
+ let state = entry.get_mut();
+ state.committed_offset =
state.committed_offset.max(committed_offset);
+ state.persisted_high_water =
state.persisted_high_water.max(persisted_high_water);
+ }
+ Entry::Vacant(entry) => {
+ entry.insert(DurableOffsetState {
+ committed_offset,
+ persisted_high_water,
+ });
+ drop(entries);
+ self.bump_membership_epoch();
+ }
+ }
+ }
+
+ pub(crate) fn remove(&self, kind: ConsumerKind, id: u32) -> bool {
+ let removed = self.entries(kind).borrow_mut().remove(&id).is_some();
+ if removed {
+ self.bump_membership_epoch();
+ }
+ removed
+ }
+
+ pub(crate) fn clear(&self) {
+ self.consumers.borrow_mut().clear();
+ self.groups.borrow_mut().clear();
+ self.bump_membership_epoch();
+ }
+
+ #[cfg(any(test, feature = "simulator"))]
+ pub(crate) fn committed_entries(&self, kind: ConsumerKind) -> Vec<(u32,
u64)> {
+ self.entries(kind)
+ .borrow()
+ .iter()
+ .map(|(id, state)| (*id, state.committed_offset))
+ .collect()
+ }
+
+ pub(crate) fn snapshot_entries(
+ &self,
+ kind: ConsumerKind,
+ ) -> Ref<'_, HashMap<u32, DurableOffsetState>> {
+ self.entries(kind).borrow()
+ }
+
+ const fn entries(&self, kind: ConsumerKind) -> &RefCell<HashMap<u32,
DurableOffsetState>> {
+ match kind {
+ ConsumerKind::Consumer => &self.consumers,
+ ConsumerKind::ConsumerGroup => &self.groups,
+ }
+ }
+
+ fn bump_membership_epoch(&self) {
+ self.membership_epoch
+ .set(self.membership_epoch.get().wrapping_add(1));
+ }
+}
+
+#[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, Arc<ProvisionalToken>>>,
+ stranded: RefCell<HashSet<u32>>,
+ uncertain: Cell<bool>,
+ durable_warned: Cell<bool>,
+ map_warned: Cell<bool>,
+ reclaim_epoch: Arc<AtomicU64>,
+ last_reclaim: Cell<Option<(u64, u64)>>,
+}
+
+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),
+ reclaim_epoch: Arc::new(AtomicU64::new(0)),
+ last_reclaim: Cell::new(None),
+ }
+ }
+
+ 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> {
+ if self.holds(id, durable) || self.stranded.borrow().contains(&id) {
+ return Ok(());
+ }
+ let limit = self.limit.get();
+ let durable_count = durable.count(self.kind);
+ // A full durable table cannot gain room by pruning provisional keys.
+ let occupied = if durable_count >= limit {
+ durable_count
+ } else {
+ let mut provisional = self.provisional.borrow_mut();
+ if provisional.len() >= limit {
+ provisional
+ .retain(|key, token| *key == id ||
token.active.load(Ordering::Relaxed) > 0);
+ }
+ drop(provisional);
+ self.occupied(durable)
+ };
+ 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(),
+ });
+ }
+ self.durable_warned.set(false);
+ 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 = Arc::clone(provisional.entry(id).or_insert_with(|| {
+ Arc::new(ProvisionalToken {
+ reclaim_epoch: Arc::clone(&self.reclaim_epoch),
+ active: AtomicUsize::new(0),
+ })
+ }));
+ token.active.fetch_add(1, Ordering::Relaxed);
+ 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| Arc::ptr_eq(token, &reservation.token))
+ }
+
+ pub(crate) fn holds(&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.active.load(Ordering::Relaxed) > 0)
+ }
+
+ pub(crate) fn set_pending_count(&self, id: u32, count: usize) {
+ if count == 0 {
+ if self.pending.borrow_mut().remove(&id).is_some() {
+ self.note_local_key_change();
+ }
+ } 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);
+ self.note_local_key_change();
+ } 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.note_local_key_change();
+ 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);
+ self.note_local_key_change();
+ }
+
+ 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 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()
+ || durable.count(self.kind) >= self.limit.get()
+ {
+ return;
+ }
+ if self.occupied(durable) < self.limit.get() {
+ self.durable_warned.set(false);
+ }
+ }
+
+ pub(crate) const fn admit_local_map_key(
+ &self,
+ map_len: usize,
+ durable_full: bool,
+ ) -> Result<(), ConsumerOffsetCapacityError> {
+ let limit = self.limit.get();
+ if map_len < limit {
+ return Ok(());
+ }
+ Err(ConsumerOffsetCapacityError {
+ kind: self.kind,
+ occupied: map_len,
+ limit,
+ first_in_episode: !self.map_warned.replace(true),
+ uncertain: !durable_full,
+ })
+ }
+
+ pub(crate) fn rearm_map_if_below_limit(&self, map_len: usize) {
+ if map_len < self.limit.get() {
+ self.map_warned.set(false);
+ }
+ }
+
+ pub(crate) fn note_local_key_change(&self) {
+ self.reclaim_epoch.fetch_add(1, Ordering::Relaxed);
+ }
+
+ pub(crate) fn forget_inactive_provisional(&self, id: u32) {
+ let mut provisional = self.provisional.borrow_mut();
+ if provisional
+ .get(&id)
+ .is_some_and(|token| token.active.load(Ordering::Relaxed) == 0)
+ {
+ provisional.remove(&id);
+ }
+ }
+
+ pub(crate) fn should_reclaim(&self, durable: &DurableConsumerOffsets) ->
bool {
+ if self.uncertain.get() {
+ return false;
+ }
+ let epoch = (
+ self.reclaim_epoch.load(Ordering::Relaxed),
+ durable.membership_epoch.get(),
+ );
+ self.last_reclaim.replace(Some(epoch)) != Some(epoch)
+ }
+
+ pub(crate) fn occupied(&self, durable: &DurableConsumerOffsets) -> usize {
Review Comment:
simplification: the three-way chain is a set union, so one `HashSet` pass
minus durable members is about 14 lines shorter. only worth it alongside the
admission fast path, since it allocates where this does not.
--
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]
