hubcio commented on code in PR #4063:
URL: https://github.com/apache/iggy/pull/4063#discussion_r3940241746
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2813,6 +2945,15 @@ where
}
}
+ if !offsets_written {
Review Comment:
warning: one failed offset write or dir fsync now fails the install after
the journal is cleared and old files unlinked, so converge wipes the segments
and the shard re-pulls from the next peer. retry the writes a few times first,
like the segment open at line 2632, converge steps at 3119 and 3122 included.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -1361,9 +1366,9 @@ impl IggyConsumer {
if is_consumer_group {
joined_consumer_group.store(false, ORDERING);
}
- trace!("Retrying to poll messages in {retry_interval}...");
- sleep(retry_interval.get_duration()).await;
}
+ trace!("Retrying to poll messages in {retry_interval}...");
+ sleep(retry_interval.get_duration()).await;
Review Comment:
warning: this sleep used to sit inside the disconnect/unauthenticated/stale
arm, now every poll error waits `polling_retry_interval` before it surfaces,
and the docs at 469 and 551 still describe the old scope. scope it to the
retryable errors, or keep it and update both docs and the PR body.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -395,13 +447,15 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>(
operation = ?header.operation,
"partition request with unresolved namespace; replying denied"
);
- send_deny_reply(
- shard,
- transport_client_id,
- &header,
- IggyError::ResourceNotFound(String::new()).as_code(),
- )
- .await;
+ let status = if matches!(
Review Comment:
warning: only the two group-not-found codes get through,
`ConsumerGroupPartitionNotOwned` and `InvalidIdentifier` from the same fence
still collapse to `ResourceNotFound` on store and delete, while the poll path
types them. forward `error.as_code()` for every resolver error.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -960,47 +994,114 @@ async fn tear_down_owned_partition(
counters.removed_local += 1;
}
-/// Reclaim consumer-group offsets left behind by a `DeleteConsumerGroup` whose
-/// topic still exists (a topic/stream delete already drops the whole partition
-/// directory, offsets included). For each owned partition, any stored
-/// consumer-group offset whose group id is no longer present in the topic's
-/// committed metadata is removed (in-memory entry + persisted file).
Monotonic,
-/// never-reused group ids make this purely reclamation -- a recreated group
-/// gets a fresh id and never reads a dead group's offset -- so it is safe to
do
-/// lazily on the reconcile pass rather than synchronously on delete.
-async fn reconcile_consumer_group_offsets(ctx: &ReconcilerCtx, counters: &mut
PassCounters) {
+/// Reclaim deleted groups through ordered offset deletes. Replicas must see
+/// each delete before a replacement store can reuse its durable slot.
+fn reconcile_consumer_group_offsets(ctx: &ReconcilerCtx, counters: &mut
PassCounters) {
let live_groups = snapshot_topic_live_groups(ctx);
let partitions = ctx.shard.plane.partitions();
let owned: Vec<IggyNamespace> = partitions.namespaces().copied().collect();
- for ns in owned {
- let live = live_groups.get(&(ns.stream_id(), ns.topic_id()));
- // Take the in-memory removes + owned unlink paths under a
closure-scoped
- // borrow that cannot escape into the await below. Holding a raw
- // `&IggyPartition` across `delete_persisted_offset().await` would let
the
- // pump task realloc the partitions vec underneath us (a UAF).
- let paths = partitions.with_partition(&ns, |partition| {
- partition.reclaim_dead_group_offsets(|group_id| {
- live.is_some_and(|set| set.contains(&group_id))
- })
- });
- let Some(paths) = paths else {
+ for namespace in owned {
+ if ctx
+ .group_offset_cleanup_inflight
+ .borrow()
+ .contains(&namespace)
+ {
+ continue;
+ }
+ let Some((next_group_id, live)) =
+ live_groups.get(&(namespace.stream_id(), namespace.topic_id()))
+ else {
continue;
};
- for path in paths {
- if let Err(err) = delete_persisted_offset(&path).await {
- warn!(
- shard = ctx.shard.id,
- ns_raw = ns.inner(),
- error = %err,
- "reconciler failed to reclaim deleted consumer-group
offset"
- );
- continue;
+ let dead = partitions
+ .with_partition(&namespace, |partition| {
+ partition.dead_consumer_group_offset_ids(|group_id| {
+ // A lagging metadata replica cannot prove a group deleted
+ // until it has applied the allocation of that group's id.
+ group_id >= *next_group_id || live.contains(&group_id)
+ })
+ })
+ .unwrap_or_default();
+ // Bound work per pass so a historical directory cannot monopolize the
+ // reconciler. Unprocessed keys keep the partition's dirty flag armed.
+ let mut tickets =
Vec::with_capacity(dead.len().min(GROUP_OFFSET_DELETES_PER_PASS));
+ for consumer_id in
dead.into_iter().take(GROUP_OFFSET_DELETES_PER_PASS) {
Review Comment:
warning: a committed delete for a stranded id whose unlink fails again
(numeric dir, `chattr +i`, EROFS) ends in `FatalCommit`, and boot replays the
op, so the replica stays down until someone fixes the fs. treat unlink failure
on a committed delete as strand-and-ok, and skip ids that already failed or
this turns into a delete loop.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -4320,13 +4859,13 @@ where
let committed_batch_stats =
self.resolve_committed_visible_offsets(&drained);
let mut messages_committed = false;
- for (mut entry, batch_stats) in
drained.into_iter().zip(committed_batch_stats) {
+ for (entry, batch_stats) in drained.iter().zip(&committed_batch_stats)
{
Review Comment:
nit: the walk is now apply-all then advance-and-reply-all, so a
`FatalCommit` at entry k leaves k-1 ops applied but never advanced, waiters
dropped, dedup fold at 4935 skipped, and the flush failure at 4913 blames
`drained.last()` instead of the failing op. comment the widened window, or
advance and reply per entry.
##########
core/partitions/src/consumer_offset_capacity.rs:
##########
@@ -0,0 +1,615 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use iggy_common::ConsumerKind;
+use std::cell::{Cell, RefCell};
+use std::collections::{HashMap, HashSet};
+use std::rc::Rc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Weak};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DurableOffsetState {
+ pub(crate) committed_offset: u64,
+ pub(crate) persisted_high_water: Option<u64>,
+}
+
+#[derive(Debug, Default)]
+pub struct DurableConsumerOffsets {
+ consumers: RefCell<HashMap<u32, DurableOffsetState>>,
+ groups: RefCell<HashMap<u32, DurableOffsetState>>,
+ 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
+ .is_some_and(|persisted| persisted >= offset)
+ })
+ }
+
+ pub(crate) fn record_explicit(
+ &self,
+ kind: ConsumerKind,
+ id: u32,
+ committed_offset: u64,
+ persisted_high_water: Option<u64>,
+ ) {
+ self.entries(kind).borrow_mut().insert(
+ id,
+ DurableOffsetState {
+ committed_offset,
+ persisted_high_water,
+ },
+ );
+ }
+
+ pub(crate) fn record_auto_commit(
+ &self,
+ kind: ConsumerKind,
+ id: u32,
+ committed_offset: u64,
+ persisted_high_water: u64,
+ ) {
+ let mut entries = self.entries(kind).borrow_mut();
+ let state = entries.entry(id).or_insert(DurableOffsetState {
+ committed_offset,
+ persisted_high_water: None,
+ });
+ state.committed_offset = state.committed_offset.max(committed_offset);
+ state.persisted_high_water = Some(
+ state
+ .persisted_high_water
+ .unwrap_or(0)
+ .max(persisted_high_water),
+ );
+ }
+
+ pub(crate) fn mark_persisted(&self, kind: ConsumerKind, id: u32,
high_water: u64) {
+ if let Some(state) = self.entries(kind).borrow_mut().get_mut(&id) {
+ state.persisted_high_water = Some(high_water);
+ }
+ }
+
+ pub(crate) fn remove(&self, kind: ConsumerKind, id: u32) -> bool {
+ let removed = self.entries(kind).borrow_mut().remove(&id).is_some();
+ if removed {
+ self.membership_epoch
+ .set(self.membership_epoch.get().wrapping_add(1));
+ }
+ removed
+ }
+
+ pub(crate) fn clear(&self) {
+ self.consumers.borrow_mut().clear();
+ self.groups.borrow_mut().clear();
+ self.membership_epoch
+ .set(self.membership_epoch.get().wrapping_add(1));
+ }
+
+ pub(crate) fn committed_entries(&self, kind: ConsumerKind) -> Vec<(u32,
u64)> {
+ self.entries(kind)
+ .borrow()
+ .iter()
+ .map(|(id, state)| (*id, state.committed_offset))
+ .collect()
+ }
+
+ const fn entries(&self, kind: ConsumerKind) -> &RefCell<HashMap<u32,
DurableOffsetState>> {
+ match kind {
+ ConsumerKind::Consumer => &self.consumers,
+ ConsumerKind::ConsumerGroup => &self.groups,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ConsumerOffsetCapacityError {
+ pub kind: ConsumerKind,
+ pub occupied: usize,
+ pub limit: usize,
+ pub first_in_episode: bool,
+ pub uncertain: bool,
+}
+
+impl From<ConsumerOffsetCapacityError> for iggy_common::IggyError {
+ fn from(error: ConsumerOffsetCapacityError) -> Self {
+ if error.uncertain {
+ Self::TransientNotAccepted
+ } else {
+ Self::TooManyConsumerOffsets
+ }
+ }
+}
+
+#[derive(Debug)]
+pub struct ConsumerOffsetCapacity {
+ kind: ConsumerKind,
+ limit: Cell<usize>,
+ pending: RefCell<HashMap<u32, usize>>,
+ provisional: RefCell<HashMap<u32, Weak<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 durable.contains(self.kind, id)
+ || self.pending.borrow().contains_key(&id)
+ || self
+ .provisional
+ .borrow()
+ .get(&id)
+ .is_some_and(|token| token.strong_count() > 0)
+ || self.stranded.borrow().contains(&id)
+ {
+ return Ok(());
+ }
+ 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 {
+ self.provisional
+ .borrow_mut()
+ .retain(|_, token| token.strong_count() > 0);
+ 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 = provisional
+ .get(&id)
+ .and_then(Weak::upgrade)
+ .unwrap_or_else(|| {
+ let token = Arc::new(ProvisionalToken {
Review Comment:
nit: this mints a fresh `Arc<ProvisionalToken>` per auto-commit poll past
durable, and the token is dead again by the next poll, so it is a malloc and
free per poll. a strong `Arc` in the map fixes it only if the reclaim-epoch
bump moves out of `ProvisionalToken::drop` into a `Drop` for the reservation,
or reclaim stops firing.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -1919,53 +1968,93 @@ where
self.transfer_offer_cache.borrow_mut().take();
}
- /// Snapshot the live offset maps + purge generation into the wire shape.
- /// Eagerly auto-committed offsets can run slightly ahead of committed
- /// state; that is safe because their covering ops sit in
- /// `(commit_op, commit_max]`, which the receiver's tail repair replays,
- /// and offset applies converge (monotone auto-commit, verbatim stores).
- fn offsets_wire_snapshot(&self) -> ConsumerOffsetsWire {
- // Every key is minted from a u32 wire id, so the narrowing filter is
- // an invariant, not a policy: say so out loud instead of silently
- // shrinking the snapshot when it ever breaks.
- let mut consumers: Vec<(u32, u64)> = self
- .consumer_offsets
- .pin()
- .iter()
- .filter_map(|(id, offset)| {
- let narrowed = u32::try_from(*id).ok();
- debug_assert!(narrowed.is_some(), "consumer offset key {id}
exceeds u32");
- narrowed.map(|id| (id, offset.offset.load(Ordering::Acquire)))
- })
- .collect();
+ fn validate_consumer_offset_transfer_counts(&self) -> Result<(),
PartitionTransferUnavailable> {
+ for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] {
+ let count = self.durable_consumer_offsets.count(kind);
+ if let Err(error) = validate_consumer_offset_transfer_count(
+ kind,
+ count,
+ CONSUMER_OFFSETS_ENTRIES_MAX as usize,
+ ) {
+ tracing::error!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.consensus().group(),
+ ?kind,
+ count,
+ max = CONSUMER_OFFSETS_ENTRIES_MAX,
+ "consumer offset state exceeds the transfer ceiling"
+ );
+ return Err(error);
+ }
+ }
+ Ok(())
+ }
+
+ /// Snapshot committed durable offsets only. Eager auto-commit progress and
+ /// follower-local cursor entries stay in the live maps until a replicated
+ /// store commits them, so neither can be promoted by state transfer.
+ fn offsets_wire_snapshot(&self) -> Result<ConsumerOffsetsWire,
PartitionTransferUnavailable> {
+ self.validate_consumer_offset_transfer_counts()?;
+ let mut consumers = self
Review Comment:
simplification: `committed_entries()` walks the map and then `durable.get()`
re-probes each id, and the 16-line body is repeated for both kinds. one
`snapshot_kind(kind, map_contains)` over the durable map does it in one pass,
it just needs a `pub(crate)` `entries` accessor.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2260,30 +2425,56 @@ where
|value| PendingConsumerOffsetCommit::upsert(kind, consumer_id,
value),
);
- if let Err(error) = self.persist_consumer_offset_commit(pending).await
{
- 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()),
- );
- return;
+ let persisted = async {
+ self.persist_consumer_offset_commit(pending).await?;
+ self.flush_consumer_offset_directories().await
}
- if let Err(error) = self.apply_consumer_offset_commit(pending) {
+ .await;
+ if let Err(error) = persisted {
+ if offset.is_some() {
+ self.release_consumer_offset_reservation(kind, consumer_id);
+ if !self.durable_consumer_offsets.contains(kind, consumer_id)
+ && let Some(path) = self.persisted_offset_path(kind,
consumer_id)
+ {
+ if delete_persisted_offset(&path).await.is_ok() {
+ self.consumer_offset_capacity_for(kind)
+ .clear_stranded(consumer_id);
+ } else {
+ self.consumer_offset_capacity_for(kind)
+ .record_stranded(consumer_id);
+ }
+ }
+ }
+ if offset.is_none() {
Review Comment:
simplification: `if offset.is_some() {..}` followed by `if offset.is_none()
{..}` is an if/else. make it `} else {`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2421,6 +2830,29 @@ where
}
};
+ if args.auto_commit
+ && let Ok(pending) =
PendingConsumerOffsetCommit::try_from_polling_consumer(consumer, 0)
+ {
+ let capacity = self.consumer_offset_capacity_for(pending.kind);
+ if !capacity.is_uncertain()
+ && self.consumer_offset_map_count(pending.kind) >=
capacity.limit()
Review Comment:
warning: `consumer_offset_map_count()` is papaya `len()`, a sum over one
counter per cpu, and it runs on every auto-commit poll before we know the key
is absent. check `contains_key` first and only call `len()` for absent keys.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2959,6 +3100,30 @@ 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 dir in self
+ .consumer_offsets_path
+ .iter()
+ .chain(self.consumer_group_offsets_path.iter())
+ {
+ for path in strayed_offset_files(Some(dir), &[]) {
Review Comment:
warning: converge unlinks the file but never calls `clear_stranded`, so an
id the install failed to unlink keeps a capacity slot until restart and
standalone consumer ids are never reclaimed. call
`clear_stranded(numeric_offset_id(path))` per unlinked path, like 2826 and the
purge path do.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2355,14 +2635,143 @@ 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))
+ }
+
+ #[allow(clippy::too_many_lines)]
+ fn resynchronize_consumer_offset_reservations(&mut self) {
let current_view = self.consensus.view();
- if current_view == self.observed_view {
+ let scan_state = (
+ self.consensus.commit_min(),
+ self.consensus.commit_max(),
+ self.consensus.sequencer().current_sequence(),
+ self.log.journal().inner.last_op(),
+ );
+ let retry_uncertain = (self.consumer_offset_capacity.is_uncertain()
+ || self.consumer_group_offset_capacity.is_uncertain())
+ && self.offset_reservations_scan_state != Some(scan_state);
Review Comment:
warning: while a capacity is uncertain this retries on every op, and each
retry is a full `repair_headers_in` scan of the headers and the evicted ring,
on the replica that is already behind. retry once per consensus tick or when
`last_op`/`commit_max` cross the failed scan's `to_op`. the comment at 5311
saying this never runs per commit is now stale.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2355,14 +2635,143 @@ 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))
+ }
+
+ #[allow(clippy::too_many_lines)]
+ fn resynchronize_consumer_offset_reservations(&mut self) {
let current_view = self.consensus.view();
- if current_view == self.observed_view {
+ let scan_state = (
+ self.consensus.commit_min(),
+ self.consensus.commit_max(),
+ self.consensus.sequencer().current_sequence(),
+ self.log.journal().inner.last_op(),
+ );
+ let retry_uncertain = (self.consumer_offset_capacity.is_uncertain()
+ || self.consumer_group_offset_capacity.is_uncertain())
+ && self.offset_reservations_scan_state != Some(scan_state);
+ if current_view == self.observed_view
+ && !self.offset_reservations_need_resync.get()
+ && !retry_uncertain
+ {
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.consumer_group_offsets_need_reconcile.set(true);
+ self.offset_reservations_scan_state = Some(scan_state);
+ // Retry uncertainty on journal or frontier progress, not every tick.
+ self.offset_reservations_need_resync.set(false);
+ if !decode_failed {
+ self.reclaim_phantom_offsets(ConsumerKind::Consumer);
+ self.reclaim_phantom_offsets(ConsumerKind::ConsumerGroup);
+ }
+ }
+
+ fn reclaim_phantom_offsets(&self, kind: ConsumerKind) {
+ 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 keep = |id| capacity.protects(id, &self.durable_consumer_offsets);
+ match kind {
+ ConsumerKind::Consumer => self
Review Comment:
warning: this retain drops every unprotected follower-local cursor at once
when one new consumer hits a full map, and each evicted consumer restarts at 0
on its next poll. evict only enough to admit the new key, or deny it with the
transient error.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -464,7 +480,21 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool {
if ctx.last_revision.get() == Some(revision)
&& ctx.last_pass_noop.get()
&& ctx.failure_state.borrow().is_empty()
+ && ctx.group_offset_cleanup_completed.get() == 0
&& !ctx.shard.has_parked_partition_frames()
+ && !ctx.shard.plane.partitions().namespaces().any(|namespace| {
Review Comment:
nit: this scan is O(owned partitions) on every wake, the 1s tick and every
metadata commit tick, where the old guard was four O(1) reads. a shard-level
dirty flag armed where `consumer_group_offsets_need_reconcile` is set keeps the
skip O(1), it just has to honour the inflight exclusion at 491.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2260,30 +2425,56 @@ where
|value| PendingConsumerOffsetCommit::upsert(kind, consumer_id,
value),
);
- if let Err(error) = self.persist_consumer_offset_commit(pending).await
{
- 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()),
- );
- return;
+ let persisted = async {
Review Comment:
warning: persist and flush are folded into one result, so a flush failure
after a successful store skips the in-memory apply at 2468 and
`offsets_wire_snapshot` then answers `ConsumerOffsetStateInconsistent`, which
is non-transient - one dead consumer blocks state transfer from this replica.
keep the two results apart and apply the commit whenever persist succeeded.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2008,54 +2118,100 @@ where
// durably stored; the in-memory update is idempotent on replay
// because we look up by (kind, id).
self.persist_consumer_offset_commit(pending).await?;
- self.apply_consumer_offset_commit(pending)?;
+ self.apply_consumer_offset_commit(pending);
self.pending_consumer_offset_commits.remove(&op);
+ self.refresh_consumer_offset_reservation(pending.kind,
pending.consumer_id);
Ok(())
}
async fn persist_consumer_offset_commit(
&self,
pending: PendingConsumerOffsetCommit,
) -> Result<(), IggyError> {
- let Some(path) = self.persisted_offset_path(pending.kind,
pending.consumer_id) else {
- return Ok(());
- };
- let key = (pending.kind, pending.consumer_id);
+ let path = self.persisted_offset_path(pending.kind,
pending.consumer_id);
+ let capacity = self.consumer_offset_capacity_for(pending.kind);
+ let creates_group = pending.kind == ConsumerKind::ConsumerGroup
Review Comment:
nit: `creates_group` probes durable and then `record_*` probes the same key
again, and the auto-commit arm already has the answer in `tracked` at 2149.
derive it from `tracked.is_none()` there and have `record_explicit` return
`created` for the explicit arm.
also at line 2256.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2798,10 +3238,27 @@ where
message: Message<RoutedRequestHeader>,
reply: Option<consensus::Sender<Message<ReplyHeader>>>,
) {
+ self.on_request_with_reservation(message, reply, None).await;
+ }
+
+ #[allow(clippy::too_many_lines)]
+ pub(crate) async fn on_request_with_reservation(
+ &mut self,
+ message: Message<RoutedRequestHeader>,
+ reply: Option<consensus::Sender<Message<ReplyHeader>>>,
+ mut reservation: Option<crate::AutoCommitReservation>,
+ ) {
+ if reservation.as_ref().is_some_and(|reservation| {
Review Comment:
nit: this drops the auto-commit frame with a bare return, no log, so a
stale-partition drop is invisible. add a `debug!` here and at the
tombstoned/missing drops in iggy_partitions.rs:638.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2260,30 +2425,56 @@ where
|value| PendingConsumerOffsetCommit::upsert(kind, consumer_id,
value),
);
- if let Err(error) = self.persist_consumer_offset_commit(pending).await
{
- 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()),
- );
- return;
+ let persisted = async {
+ self.persist_consumer_offset_commit(pending).await?;
+ self.flush_consumer_offset_directories().await
}
- if let Err(error) = self.apply_consumer_offset_commit(pending) {
+ .await;
+ if let Err(error) = persisted {
+ if offset.is_some() {
+ self.release_consumer_offset_reservation(kind, consumer_id);
+ if !self.durable_consumer_offsets.contains(kind, consumer_id)
Review Comment:
warning: `persist_offset` truncates then writes, so a failed NoAck store on
a durable key leaves a 0-byte file, this guard skips cleanup for durable keys,
and the next boot reads it as torn and loses the committed position.
pre-existing, but this branch is new: temp-and-rename in `persist_offset`, or
reset the persisted state here so the next commit blind-writes.
##########
core/common/src/traits/message_client.rs:
##########
@@ -31,6 +31,10 @@ pub trait MessageClient {
/// Polling a consumer group the client is not (or no longer) a member of
fails with `ConsumerGroupMemberNotFound` rather than returning an empty batch,
so the caller can rejoin.
/// A member that holds no partitions gets an empty batch whose
`partition_id` is [`NO_ASSIGNED_PARTITION`](crate::NO_ASSIGNED_PARTITION).
#[allow(clippy::too_many_arguments)]
+ /// With server-side auto-commit enabled, a new consumer offset key can be
Review Comment:
nit: this paragraph sits after the `#[allow]` with no blank `///`, so
rustdoc glues it onto the previous paragraph. move it above the attribute and
add a separator.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -618,6 +618,11 @@ unsafe impl Sync for IggyConsumer {}
/// [`topic()`]: crate::prelude::IggyConsumerBuilder::topic
/// [`without_encryptor()`]:
crate::prelude::IggyConsumerBuilder::without_encryptor
/// [`without_poll_interval()`]:
crate::prelude::IggyConsumerBuilder::without_poll_interval
+///
+/// A server-side auto-commit poll can fail with `TooManyConsumerOffsets` when
Review Comment:
warning: true for the raw poll only; every `AutoCommit` variant except
`Disabled` commits the same key client-side, and a 3024 there is logged at 237
and dropped at 1075 and 1102, so the stream keeps yielding with no durable
progress. say only `AutoCommit::Disabled` avoids the key and that client-side
commits fail the same way, logged not surfaced.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2075,16 +2231,35 @@ where
consumer_id: u32,
Review Comment:
simplification: `is_auto_commit_offset_covered` has no production caller
left now that dispatch uses `reserve_durable`, only tests. delete it and assert
on `durable_consumer_offsets.covers` in the tests.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -111,14 +111,27 @@ where
spawn_poll_io(Rc::clone(&shard), namespace, plan,
reply);
}
Some(plan) => {
- let (fragments, current_offset, auto_commit) =
plan.execute_resident();
- if let Some(applied) = auto_commit {
- submit_auto_commit(&shard, namespace, &applied);
- }
- let _ = reply.try_send(PartitionReadReply::Poll {
- fragments,
- current_offset,
- });
+ let result = match plan.execute_resident() {
+ Ok((fragments, current_offset, auto_commit)) => {
+ if let Some(applied) = auto_commit
+ && let Err(error) =
+ submit_auto_commit(&shard, namespace,
&applied)
+ {
+ PartitionReadReply::Rejected(error)
+ } else {
+ PartitionReadReply::Poll {
+ fragments,
+ current_offset,
+ }
+ }
+ }
+ Err(error) => {
+
shard.metrics().record_consumer_offset_denied(error.kind);
Review Comment:
nit: this counts `uncertain: true` denials too, which the client sees as
`TransientNotAccepted`, while the help text says admission limit and the relay
at 633 already gates on the permanent variant. gate on `!error.uncertain`.
also at lines 229, 296.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -569,6 +624,17 @@ async fn relay_partition_reply<B, MJ, S, SB>(
// commits moments later. The client's read-timeout is the
recovery.
return;
};
+ if reply
+ .as_slice()
+ .get(..size_of::<iggy_binary_protocol::ReplyHeader>())
+ .and_then(|bytes| {
+
bytemuck::checked::try_from_bytes::<iggy_binary_protocol::ReplyHeader>(bytes).ok()
Review Comment:
nit: the `ReplyHeader` parse runs before `consumer_kind` is checked, so
every relayed write reply pays it though only `StoreConsumerOffset` has a kind.
test `consumer_kind` first.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -3137,9 +3614,28 @@ where
/// is bypassed at view-change reset.
#[allow(clippy::future_not_send)]
pub async fn drain_request_queue_into_prepares(&mut self, slots_freed:
usize) {
- for _ in 0..slots_freed {
+ self.resynchronize_consumer_offset_reservations();
+ let mut promoted = 0;
+ while promoted < slots_freed {
let req = self.consensus().pop_queued_request();
let Some(mut req) = req else { break };
+ let _reservation =
Review Comment:
simplification: the store body is parsed here and again at 3651, two
`decode_from` and two name hashes per promoted store. parse once above the
preflight, gated on `StoreConsumerOffset`, and reuse `(kind, id, offset)` in
both blocks.
##########
core/server/config.toml:
##########
@@ -1024,6 +1024,22 @@ prepare_queue_depth = 32
# actually sees, not the node's client total.
dedup_clients_max = 4096
+# Distinct durable consumer-offset keys a partition primary admits per kind.
+# Standalone consumers and consumer groups are counted separately. Existing
+# keys remain writable at the limit. A new key is rejected before consensus.
+# A non-empty auto_commit poll that needs a new offset key is also rejected
+# with TooManyConsumerOffsets and returns no messages. Polls with auto_commit
+# disabled do not allocate offset keys and remain available.
+# UPGRADE: existing files are loaded even above this limit, but new keys then
+# remain blocked until offsets are explicitly deleted or this limit is raised.
+# Standalone offsets have no automatic expiry. Reuse stable consumer ids and
+# count existing numeric offset files per partition and kind before upgrading.
+# The environment override is IGGY_PARTITION_CONSUMER_OFFSETS_MAX.
+# On replicated partitions, offset stores and deletes carrying NoAck now wait
Review Comment:
nit: "now" is a changelog tell in shipped config, and the UPGRADE line above
covers the quota, not NoAck. drop "now", and give the NoAck change its own
UPGRADE line if operators need the before and after.
##########
core/common/src/error/iggy_error.rs:
##########
@@ -333,6 +333,8 @@ pub enum IggyError {
NotResolvedConsumer(Identifier) = 3022,
#[error("Cannot open consumer offsets file for path: {0}")]
CannotOpenConsumerOffsetsFile(String) = 3023,
+ #[error("Per-partition consumer offset limit reached (see [partition]
consumer_offsets_max)")]
Review Comment:
nit: only error string here that embeds a server config key, mirrored into
the go and node tables, so SDK users get an operator hint and a reword is four
edits. use "Consumer offset limit reached for partition" and keep the config
pointer in the server warn.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2008,54 +2118,100 @@ where
// durably stored; the in-memory update is idempotent on replay
// because we look up by (kind, id).
self.persist_consumer_offset_commit(pending).await?;
- self.apply_consumer_offset_commit(pending)?;
+ self.apply_consumer_offset_commit(pending);
self.pending_consumer_offset_commits.remove(&op);
+ self.refresh_consumer_offset_reservation(pending.kind,
pending.consumer_id);
Ok(())
}
async fn persist_consumer_offset_commit(
&self,
pending: PendingConsumerOffsetCommit,
) -> Result<(), IggyError> {
- let Some(path) = self.persisted_offset_path(pending.kind,
pending.consumer_id) else {
- return Ok(());
- };
- let key = (pending.kind, pending.consumer_id);
+ let path = self.persisted_offset_path(pending.kind,
pending.consumer_id);
+ let capacity = self.consumer_offset_capacity_for(pending.kind);
+ let creates_group = pending.kind == ConsumerKind::ConsumerGroup
+ && !self
+ .durable_consumer_offsets
+ .contains(pending.kind, pending.consumer_id);
match pending.mutation {
// A server auto-commit persists monotonically: its op offset can
// trail the durably-recorded value (disk-tier polls replicate in
// IO-completion order), so a plain overwrite would rewind the file
- // and re-deliver on restart. The `persisted_offsets` tracker keeps
+ // and re-deliver on restart. The durable offset tracker keeps
// the fold off the file: a covered offset skips the write, an
// advancing one blind-writes, and only a cold key (first commit
// after boot) reads the file once. Explicit client stores
// overwrite, so a deliberate offset reset still holds. Mirrors the
// in-memory `upsert_offset_max` vs `upsert_offset` split in the
// commit-apply.
PendingConsumerOffsetMutation::Upsert(offset) if
pending.auto_commit => {
- let tracked =
self.persisted_offsets.borrow().get(&key).copied();
- let persisted = match tracked {
- Some(high_water) if offset <= high_water => return Ok(()),
- Some(_) => {
- persist_offset(&path, offset,
self.consumer_offset_enforce_fsync).await?;
- offset
+ let tracked = self
+ .durable_consumer_offsets
+ .get(pending.kind, pending.consumer_id);
+ let persisted_high_water = match (path.as_deref(), tracked) {
+ (None, _) => offset,
+ (Some(_), Some(state))
+ if state
+ .persisted_high_water
+ .is_some_and(|high_water| offset <= high_water) =>
+ {
+ state.persisted_high_water.expect("covered persisted
value")
+ }
+ (Some(path), Some(state)) => {
+ let value = state.committed_offset.max(offset);
+ persist_offset(path, value,
self.consumer_offset_enforce_fsync).await?;
+ value
}
- None => {
- persist_offset_max(&path, offset,
self.consumer_offset_enforce_fsync)
- .await?
+ (Some(path), None) => {
+ persist_offset_max(path, offset,
self.consumer_offset_enforce_fsync).await?
}
};
- self.persisted_offsets.borrow_mut().insert(key, persisted);
+ self.durable_consumer_offsets.record_auto_commit(
+ pending.kind,
+ pending.consumer_id,
+ if tracked.is_none() {
+ persisted_high_water
+ } else {
+ offset
+ },
+ persisted_high_water,
+ );
+ capacity.clear_stranded(pending.consumer_id);
+ if creates_group {
+ self.consumer_group_offsets_need_reconcile.set(true);
+ }
Ok(())
}
PendingConsumerOffsetMutation::Upsert(offset) => {
- persist_offset(&path, offset,
self.consumer_offset_enforce_fsync).await?;
- self.persisted_offsets.borrow_mut().insert(key, offset);
+ if let Some(path) = path.as_deref() {
+ persist_offset(path, offset,
self.consumer_offset_enforce_fsync).await?;
+ }
+ self.durable_consumer_offsets.record_explicit(
+ pending.kind,
+ pending.consumer_id,
+ offset,
+ Some(offset),
+ );
+ capacity.clear_stranded(pending.consumer_id);
+ if creates_group {
+ self.consumer_group_offsets_need_reconcile.set(true);
+ }
Ok(())
}
PendingConsumerOffsetMutation::Delete => {
- delete_persisted_offset(&path).await?;
- self.persisted_offsets.borrow_mut().remove(&key);
+ if let Some(path) = path.as_deref() {
+ delete_persisted_offset(path).await?;
+ self.consumer_offset_dirs_dirty[match pending.kind {
+ ConsumerKind::Consumer => 0,
Review Comment:
nit: a `match` inlined inside the index subscript reads badly. hoist the
index into a local.
##########
core/partitions/src/consumer_offset_capacity.rs:
##########
@@ -0,0 +1,615 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use iggy_common::ConsumerKind;
+use std::cell::{Cell, RefCell};
+use std::collections::{HashMap, HashSet};
+use std::rc::Rc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Weak};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DurableOffsetState {
+ pub(crate) committed_offset: u64,
+ pub(crate) persisted_high_water: Option<u64>,
+}
+
+#[derive(Debug, Default)]
+pub struct DurableConsumerOffsets {
+ consumers: RefCell<HashMap<u32, DurableOffsetState>>,
+ groups: RefCell<HashMap<u32, DurableOffsetState>>,
+ 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
+ .is_some_and(|persisted| persisted >= offset)
+ })
+ }
+
+ pub(crate) fn record_explicit(
+ &self,
+ kind: ConsumerKind,
+ id: u32,
+ committed_offset: u64,
+ persisted_high_water: Option<u64>,
+ ) {
+ self.entries(kind).borrow_mut().insert(
+ id,
+ DurableOffsetState {
+ committed_offset,
+ persisted_high_water,
+ },
+ );
+ }
+
+ pub(crate) fn record_auto_commit(
+ &self,
+ kind: ConsumerKind,
+ id: u32,
+ committed_offset: u64,
+ persisted_high_water: u64,
+ ) {
+ let mut entries = self.entries(kind).borrow_mut();
+ let state = entries.entry(id).or_insert(DurableOffsetState {
+ committed_offset,
+ persisted_high_water: None,
+ });
+ state.committed_offset = state.committed_offset.max(committed_offset);
+ state.persisted_high_water = Some(
+ state
+ .persisted_high_water
+ .unwrap_or(0)
+ .max(persisted_high_water),
+ );
+ }
+
+ pub(crate) fn mark_persisted(&self, kind: ConsumerKind, id: u32,
high_water: u64) {
+ if let Some(state) = self.entries(kind).borrow_mut().get_mut(&id) {
+ state.persisted_high_water = Some(high_water);
+ }
+ }
+
+ pub(crate) fn remove(&self, kind: ConsumerKind, id: u32) -> bool {
+ let removed = self.entries(kind).borrow_mut().remove(&id).is_some();
+ if removed {
+ self.membership_epoch
+ .set(self.membership_epoch.get().wrapping_add(1));
+ }
+ removed
+ }
+
+ pub(crate) fn clear(&self) {
+ self.consumers.borrow_mut().clear();
+ self.groups.borrow_mut().clear();
+ self.membership_epoch
+ .set(self.membership_epoch.get().wrapping_add(1));
+ }
+
+ pub(crate) fn committed_entries(&self, kind: ConsumerKind) -> Vec<(u32,
u64)> {
+ self.entries(kind)
+ .borrow()
+ .iter()
+ .map(|(id, state)| (*id, state.committed_offset))
+ .collect()
+ }
+
+ const fn entries(&self, kind: ConsumerKind) -> &RefCell<HashMap<u32,
DurableOffsetState>> {
+ match kind {
+ ConsumerKind::Consumer => &self.consumers,
+ ConsumerKind::ConsumerGroup => &self.groups,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ConsumerOffsetCapacityError {
+ pub kind: ConsumerKind,
+ pub occupied: usize,
+ pub limit: usize,
+ pub first_in_episode: bool,
+ pub uncertain: bool,
+}
+
+impl From<ConsumerOffsetCapacityError> for iggy_common::IggyError {
+ fn from(error: ConsumerOffsetCapacityError) -> Self {
+ if error.uncertain {
+ Self::TransientNotAccepted
+ } else {
+ Self::TooManyConsumerOffsets
+ }
+ }
+}
+
+#[derive(Debug)]
+pub struct ConsumerOffsetCapacity {
+ kind: ConsumerKind,
+ limit: Cell<usize>,
+ pending: RefCell<HashMap<u32, usize>>,
+ provisional: RefCell<HashMap<u32, Weak<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 durable.contains(self.kind, id)
+ || self.pending.borrow().contains_key(&id)
+ || self
+ .provisional
+ .borrow()
+ .get(&id)
+ .is_some_and(|token| token.strong_count() > 0)
+ || self.stranded.borrow().contains(&id)
+ {
+ return Ok(());
+ }
+ 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 {
+ self.provisional
+ .borrow_mut()
+ .retain(|_, token| token.strong_count() > 0);
+ 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 = provisional
+ .get(&id)
+ .and_then(Weak::upgrade)
+ .unwrap_or_else(|| {
+ let token = Arc::new(ProvisionalToken {
+ reclaim_epoch: Arc::clone(&self.reclaim_epoch),
+ });
+ provisional.insert(id, Arc::downgrade(&token));
+ token
+ });
+ Ok(AutoCommitReservation {
+ token,
+ kind: self.kind,
+ consumer_id: id,
+ })
+ }
+
+ pub(crate) fn owns(&self, reservation: &AutoCommitReservation) -> bool {
+ reservation.kind == self.kind
+ && self
+ .provisional
+ .borrow()
+ .get(&reservation.consumer_id)
+ .is_some_and(|token| std::ptr::eq(token.as_ptr(),
Arc::as_ptr(&reservation.token)))
+ }
+
+ pub(crate) fn protects(&self, id: u32, durable: &DurableConsumerOffsets)
-> bool {
+ durable.contains(self.kind, id)
Review Comment:
simplification: `protects` is the first arm of `check` minus `stranded`.
share a `holds(id, durable)` helper and keep the deliberate `stranded`
difference in the callers.
##########
core/server/src/offset_recovery.rs:
##########
@@ -33,13 +33,16 @@ use tracing::{error, trace, warn};
const COMPONENT: &str = "STREAMING_PARTITIONS";
-pub fn load_consumer_offsets(path: &str) -> Result<Vec<ConsumerOffset>,
IggyError> {
+pub type RecoveredOffsets<T> = (Vec<T>, Vec<u32>);
Review Comment:
simplification: the tuple alias is destructured by position in
partition_helpers.rs:190 and 231, and the `is_file()` re-stat at 89 and 164
only tells corrupt-and-removed from torn. have `read_offset_file` return
`Loaded | Removed | Stranded` and a named struct, which drops the re-stat too.
##########
core/partitions/src/consumer_offset_capacity.rs:
##########
@@ -0,0 +1,615 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use iggy_common::ConsumerKind;
+use std::cell::{Cell, RefCell};
+use std::collections::{HashMap, HashSet};
+use std::rc::Rc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Weak};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DurableOffsetState {
+ pub(crate) committed_offset: u64,
+ pub(crate) persisted_high_water: Option<u64>,
Review Comment:
simplification: `None` only exists inside `install_state_transfer` between
`record_explicit(.., None)` and `mark_persisted`, and converge clears the map
before anything can observe it, so this can be a plain `u64` and the `(Some,
Some)` arm at iggy_partition.rs:2154 becomes one compare. it drops a
defence-in-depth check, so only if the install stays fail-closed.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -111,14 +111,27 @@ where
spawn_poll_io(Rc::clone(&shard), namespace, plan,
reply);
}
Some(plan) => {
- let (fragments, current_offset, auto_commit) =
plan.execute_resident();
- if let Some(applied) = auto_commit {
- submit_auto_commit(&shard, namespace, &applied);
- }
- let _ = reply.try_send(PartitionReadReply::Poll {
- fragments,
- current_offset,
- });
+ let result = match plan.execute_resident() {
Review Comment:
simplification: this match is byte-identical to the one at 215. one
`poll_reply(shard, namespace, result)` keeps a single copy of the deny
sequence, the line saving is small but there is one place to fix.
##########
core/server/src/consumer_group.rs:
##########
@@ -226,15 +226,28 @@ where
let body = request_body(&request);
// The store/delete ops differ only in the decode type; this collapses
// their identical decode -> resolve group id -> rewrite consumer id ->
- // re-encode bodies. A non-group consumer or unresolved group returns the
- // request untouched (the apply/read path handles the miss).
+ // re-encode bodies. Individual consumers pass through. A group identifier
+ // that metadata cannot resolve is rejected before it can create a raw file
+ // in the group-offset directory.
macro_rules! rewrite_group_offset {
($ty:ty) => {{
let mut wire = <$ty>::decode_from(body).map_err(|_|
IggyError::InvalidCommand)?;
+ if wire.consumer.kind != KIND_CONSUMER_GROUP {
+ return Ok(request);
+ }
let Some(group_id) =
Review Comment:
simplification: this branch and `topic_exists` repeat `fence_group_offset`
in responses.rs:295 character for character, and the fence already runs first
on every transport. one shared classifier for both, and the comment at
dispatch/partition.rs:528 saying this is unreachable is now wrong - it is
reachable through the await at 501.
##########
core/integration/tests/server/consumer_offset_quota_vsr.rs:
##########
@@ -0,0 +1,493 @@
+// 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::prelude::*;
+use iggy_binary_protocol::codec::WireEncode;
+use iggy_binary_protocol::consensus::Operation;
+use
iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest;
+use iggy_binary_protocol::{AckLevel, WireConsumer, WireIdentifier};
+use iggy_common::store_consumer_offset::StoreConsumerOffset;
+use integration::harness::TestBinary;
+use integration::iggy_harness;
+use reqwest::StatusCode;
+use std::collections::BTreeMap;
+use std::fs;
+
+use super::http_client::HttpClient;
+use super::raw_tcp;
+
+const STREAM_NAME: &str = "consumer-offset-quota-stream";
+const TOPIC_NAME: &str = "consumer-offset-quota-topic";
+const PARTITION_ID: u32 = 0;
+const LIMIT: u32 = 4;
+
+#[iggy_harness(
+ cluster_nodes = 1,
+ server(partition.consumer_offsets_max = "4")
+)]
+async fn
given_full_consumer_offset_table_when_creating_another_should_reject_without_new_file(
+ harness: &TestHarness,
+) {
+ let client = harness.tcp_root_client().await.expect("TCP root client");
+ let stream = Identifier::named(STREAM_NAME).expect("stream identifier");
+ let topic = Identifier::named(TOPIC_NAME).expect("topic identifier");
+ let stream_details = client
+ .create_stream(STREAM_NAME)
+ .await
+ .expect("create stream");
+ let topic_details = client
+ .create_topic(
+ &stream,
+ TOPIC_NAME,
+ &TopicCreateOptions {
+ partitions_count: Some(1),
+ message_expiry: Some(IggyExpiry::NeverExpire),
+ ..TopicCreateOptions::default()
+ },
+ )
+ .await
+ .expect("create topic");
+ let mut messages = vec![
+ IggyMessage::builder()
+ .payload("offset-quota".into())
+ .build()
+ .expect("build message"),
+ ];
+ client
+ .send_messages(
+ &stream,
+ &topic,
+ &Partitioning::partition_id(PARTITION_ID),
+ &mut messages,
+ )
+ .await
+ .expect("seed non-empty partition");
+
+ client
+ .create_user(
+ "offset-poll-only",
+ "password123",
+ UserStatus::Active,
+ Some(Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(BTreeMap::from([(
+ stream_details.id as usize,
+ StreamPermissions {
+ topics: Some(BTreeMap::from([(
+ topic_details.id as usize,
+ TopicPermissions {
+ poll_messages: true,
+ ..Default::default()
+ },
+ )])),
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await
+ .expect("create a topic-scoped consumer");
+ let client = harness.tcp_new_client().await.expect("consumer TCP client");
+ client
+ .login_user("offset-poll-only", "password123")
+ .await
+ .expect("consumer login");
+
+ let first_consumer = Consumer::new(Identifier::numeric(1).unwrap());
+ let polled = client
+ .poll_messages(
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ &first_consumer,
+ &PollingStrategy::first(),
+ 1,
+ true,
+ )
+ .await
+ .expect("new auto-commit consumer fits");
+ assert_eq!(polled.messages.len(), 1);
+ let first_file = harness.server().data_path().join(format!(
+ "streams/{}/topics/{}/partitions/{PARTITION_ID}/offsets/consumers/1",
+ stream_details.id, topic_details.id
+ ));
+ let deadline = tokio::time::Instant::now() +
std::time::Duration::from_secs(10);
+ while !first_file.is_file() {
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "auto-commit never reached its file"
+ );
+ tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+ }
+ assert!(
+ client
+ .poll_messages(
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ &first_consumer,
+ &PollingStrategy::next(),
+ 1,
+ true
+ )
+ .await
+ .expect("next poll")
+ .messages
+ .is_empty()
+ );
+
+ for consumer_id in 1..=LIMIT {
+ client
+ .store_consumer_offset(
+
&Consumer::new(Identifier::numeric(consumer_id).expect("consumer identifier")),
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ 0,
+ )
+ .await
+ .expect("store offset within limit");
+ }
+
+ let rejected_consumer =
+ Consumer::new(Identifier::numeric(LIMIT + 1).expect("consumer
identifier"));
+ let rejected = client
+ .store_consumer_offset(&rejected_consumer, &stream, &topic,
Some(PARTITION_ID), 0)
+ .await;
+ assert!(
+ matches!(rejected, Err(IggyError::TooManyConsumerOffsets)),
+ "the first key above the limit must receive the typed capacity error"
+ );
+
+ client
+ .store_consumer_offset(
+ &Consumer::new(Identifier::numeric(1).expect("consumer
identifier")),
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ 0,
+ )
+ .await
+ .expect("existing key remains writable at the limit");
+
+ let poll_rejected = client
+ .poll_messages(
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ &rejected_consumer,
+ &PollingStrategy::first(),
+ 1,
+ true,
+ )
+ .await;
+ assert!(
+ matches!(poll_rejected, Err(IggyError::TooManyConsumerOffsets)),
+ "auto-commit must not return data when its new key cannot be admitted"
+ );
+ client
+ .poll_messages(
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ &rejected_consumer,
+ &PollingStrategy::first(),
+ 1,
+ false,
+ )
+ .await
+ .expect("the same poll succeeds when auto-commit is disabled");
+
+ client
+ .delete_consumer_offset(
+ &Consumer::new(Identifier::numeric(1).expect("consumer
identifier")),
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ )
+ .await
+ .expect("delete one accepted offset");
+ client
+ .store_consumer_offset(&rejected_consumer, &stream, &topic,
Some(PARTITION_ID), 0)
+ .await
+ .expect("delete releases one durable slot");
+
+ let mut raw = raw_tcp::connect(harness).await;
+ let raw_client_id = 0xC0FF_EE03;
+ let session = raw_tcp::register_root(&mut raw, raw_client_id).await;
+ let unresolved_group = StoreConsumerOffsetRequest {
+ consumer: WireConsumer::consumer_group(WireIdentifier::Numeric(999)),
+ stream_id: WireIdentifier::Numeric(stream_details.id),
+ topic_id: WireIdentifier::Numeric(topic_details.id),
+ partition_id: Some(PARTITION_ID),
+ offset: 0,
+ ack: AckLevel::Quorum,
+ }
+ .to_bytes();
+ let header = raw_tcp::request_header(
+ Operation::StoreConsumerOffset,
+ raw_client_id,
+ session,
+ 1,
+ unresolved_group.len(),
+ );
+ let (reply, _) = raw_tcp::exchange(&mut raw, &header,
&unresolved_group).await;
+ assert_eq!(
+ raw_tcp::reply_status(&reply),
+ IggyError::ConsumerGroupIdNotFound(Identifier::numeric(999).unwrap(),
topic.clone())
+ .as_code()
+ );
+
+ let offsets_dir = harness.server().data_path().join(format!(
+ "streams/{}/topics/{}/partitions/{PARTITION_ID}/offsets/consumers",
+ stream_details.id, topic_details.id
+ ));
+ let file_count = integration::harness::disk::consumer_offset_file_ids(
+ &harness.server().data_path(),
+ stream_details.id,
+ topic_details.id,
+ PARTITION_ID,
+ ConsumerKind::Consumer,
+ )
+ .expect("consumer offsets directory")
+ .len();
+ assert_eq!(file_count, LIMIT as usize);
+ let groups_dir = offsets_dir
Review Comment:
simplification: this hand-rolls what `disk::consumer_offset_file_ids(..,
ConsumerGroup)` already does nine lines up. call it with `.map_or(0, |ids|
ids.len())` to keep absent-dir as 0, and the `offsets_dir` binding at 255 goes
too.
--
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]
