hubcio commented on code in PR #4063:
URL: https://github.com/apache/iggy/pull/4063#discussion_r3947244611
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2186,38 +2346,46 @@ 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
Review Comment:
Warning: The comment claims stranded keys are skipped, but the body only
sorts and dedups. Add `&& !capacity.is_stranded(id)` to the push, or state why
a stranded key cannot be in `consumer_group_offsets`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -4367,7 +4979,26 @@ where
});
return;
}
+ }
+
+ // Commit replies and the applied frontier must follow directory
+ // durability. One sync per touched kind covers the whole walk. A sync
+ // failure is attributed to the batch boundary because one directory
+ // sync covers every delete in that kind, not one uniquely failing op.
+ if let Err(error) = self.flush_consumer_offset_directories().await {
+ if let Some(entry) = drained.last() {
+ error!(namespace_raw, %error, "consumer offset directory sync
failed after committed operations");
+ self.fatal = Some(FatalCommit {
Review Comment:
Critical: One sync failure flag covers both offset kinds, so a failed sync
of the groups directory sets `fatal` on a walk that only wrote consumer
offsets, shutting the node down. Attribute the failure per kind.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -8006,49 +9590,233 @@ mod tests {
let _ = std::fs::remove_dir_all(&dir);
}
- /// `reclaim_dead_group_offsets` must drop exactly the not-`is_live` groups
- /// from the in-memory map and hand back their owned persisted-file paths,
- /// leaving live groups untouched. The returned `Vec<String>` is what the
- /// reconciler unlinks off-borrow, so it carries no partition reference.
- ///
- /// Scope: the synchronous removal contract the off-borrow split relies on.
- /// The cross-task interleave it enables -- a pump mutating the partitions
vec
- /// while a sibling task is parked mid-await -- is covered on the
simulator's
- /// deterministic executor, against the debug borrow tripwire, by
- /// `simulator::tests::shell_detects_partition_borrow_held_across_await`
- /// (`swap_remove`) and
- /// `shell_detects_partition_borrow_held_across_a_pump_realloc` (a growing
- /// `push`, which relocates every element).
#[compio::test]
- async fn reclaim_dead_group_offsets_drops_dead_keeps_live() {
- let mut partition = test_partition();
- let group_offsets_path = "/iggy-test-cg-offsets".to_owned();
- partition.consumer_group_offsets_path =
Some(group_offsets_path.clone());
-
- let dead: u32 = 1;
- let live: u32 = 2;
+ async fn
given_dead_group_when_reclaimed_should_keep_capacity_until_replicated_delete() {
+ let (mut partition, _) = recording_partition();
partition.consumer_group_offsets.pin().insert(
- ConsumerGroupId(dead as usize),
- ConsumerOffset::new(ConsumerKind::ConsumerGroup, dead, 7,
String::new()),
+ ConsumerGroupId(7),
+ ConsumerOffset::new(ConsumerKind::ConsumerGroup, 7, 11,
String::new()),
+ );
+ partition.seed_recovered_consumer_offset(ConsumerKind::ConsumerGroup,
7, 11, 11);
+ assert_eq!(partition.dead_consumer_group_offset_ids(|_| false),
vec![7]);
+ assert_eq!(partition.consumer_group_offset_ids(), vec![7]);
+ assert_eq!(
+
partition.occupied_consumer_offset_count(ConsumerKind::ConsumerGroup),
+ 1
+ );
+ partition.stage_consumer_offset_delete(1, ConsumerKind::ConsumerGroup,
7);
+ partition
+ .apply_staged_consumer_offset_commit(1)
+ .await
+ .expect("delete commits");
+ assert_eq!(
+
partition.occupied_consumer_offset_count(ConsumerKind::ConsumerGroup),
+ 0
+ );
+ assert!(partition.consumer_group_offset_ids().is_empty());
+ }
+
+ #[compio::test]
+ async fn
given_known_stranded_group_file_when_reconciling_should_not_submit_delete_loop()
{
Review Comment:
Warning: This test passes with or without a stranded filter, since id 7 is
never inserted into `consumer_group_offsets` and the function iterates only
that map. Insert the id before seeding it as stranded.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2261,21 +2429,50 @@ 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);
+ if let Err(error) = self.flush_consumer_offset_directories().await {
Review Comment:
Warning: `apply_consumer_offset_commit` runs before this flush, so on a sync
error the client gets `CannotSyncFile` while the key is already durable and in
the live map. Reply success and log, or sync only the kind this request touched.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2355,14 +2649,193 @@ 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();
+ self.mark_consumer_group_offsets_need_reconcile();
+ }
+
+ 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.
+ // Within one view an op is assigned once, so uncommitted staging is
+ // kept too; only a view change can replace what sits at those ops, and
+ // only then is the tail decoded again.
+ let same_view = current_view == self.observed_view;
+ let mut rebuilt: HashMap<_, _> = self
+ .pending_consumer_offset_commits
+ .iter()
+ .filter(|(op, _)| **op >= from_op && (**op <= commit_max ||
same_view))
+ .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;
+ }
+ if rebuilt.contains_key(&op) {
+ 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.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 needed =
map_count.saturating_sub(capacity.limit()).saturating_add(1);
+ match kind {
+ ConsumerKind::Consumer => {
+ self.reclaim_phantom_offset_keys(&self.consumer_offsets, kind,
needed, |key| {
+ u32::try_from(*key).ok()
+ });
+ }
+ ConsumerKind::ConsumerGroup => self.reclaim_phantom_offset_keys(
+ &self.consumer_group_offsets,
+ kind,
+ needed,
+ |key| u32::try_from(key.0).ok(),
+ ),
+ }
+ }
+
+ fn reclaim_phantom_offset_keys<K: Hash + Eq>(
+ &self,
+ offsets: &papaya::HashMap<K, ConsumerOffset>,
+ kind: ConsumerKind,
+ mut remaining: usize,
+ consumer_id: impl Fn(&K) -> Option<u32>,
+ ) {
+ let capacity = self.consumer_offset_capacity_for(kind);
+ let map = offsets.pin();
+ for (key, _) in &map {
+ if let Some(id) = consumer_id(key)
+ && !capacity.holds(id, &self.durable_consumer_offsets)
+ && map.remove(key).is_some()
+ {
+ capacity.forget_inactive_provisional(id);
+ debug!(
Review Comment:
Warning: Evicting a live-map cursor with no durable backing sends that
consumer back to offset 0 on its next `Next` poll, a full redelivery. Raise
this to `warn!` and document it in the poll contract.
##########
core/sdk/src/leader_aware.rs:
##########
@@ -406,6 +406,10 @@ impl RosterWalk {
self.attempted.push(endpoint.clone());
Some(endpoint)
}
+
+ pub(crate) fn is_single_endpoint(&self) -> bool {
Review Comment:
Warning: `is_single_endpoint` also returns true for an empty roster, since
`RosterWalk::new` always seeds `attempted` with the current address. Gate on a
non-empty roster so a client whose roster discovery failed does not replay one
node.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -199,26 +196,55 @@ fn spawn_poll_io<B, MJ, S, SB>(
"slow partition poll; gather side may have timed out"
);
}
- // Fire-and-forget: the poll reply is not gated on the offset commit.
- if let Some(applied) = auto_commit {
- submit_auto_commit(&shard, namespace, &applied);
- }
- let _ = reply.try_send(PartitionReadReply::Poll {
- fragments,
- current_offset,
- });
+ let result = poll_reply(&shard, namespace, result);
+ let _ = reply.try_send(result);
});
}
+fn poll_reply<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ namespace: IggyNamespace,
+ result: Result<(PollFragments, u64, Option<AutoCommitApplied>),
ConsumerOffsetCapacityError>,
+) -> PartitionReadReply
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ match result {
+ Ok((fragments, current_offset, auto_commit)) => {
+ if let Some(applied) = auto_commit
+ && let Err(error) = submit_auto_commit(shard, namespace,
&applied)
+ {
+ PartitionReadReply::Rejected(error)
Review Comment:
Warning: A poll that fails auto-commit now returns `Rejected` and drops the
fragments already read, and inbox pressure reaches this with
`TransientNotAccepted`, not only the quota limit. Document that case alongside
`TooManyConsumerOffsets`.
##########
core/shard/src/lib.rs:
##########
@@ -9830,8 +9872,16 @@ fn rotate_sweep_to_cursor(namespaces: &mut
[IggyNamespace], cursor: Option<IggyN
namespaces.rotate_left(namespaces.partition_point(|namespace| *namespace <
cursor));
}
-/// Whether this replica holds adopted suffix HEADERS above `commit_max` whose
-/// bodies never arrived.
+fn partition_repair_fetch_to_op(
+ commit_min: u64,
+ commit_max: u64,
+ missing_suffix: Option<u64>,
Review Comment:
Warning: Returning the adopted suffix head while `commit_min < commit_max`
makes the repair request span the committed prefix this replica already holds.
This VSR change is unrelated to consumer offset quotas; split it into its own
PR.
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -1323,6 +1327,15 @@ impl TcpClient {
// hops would bounce the request between two nodes and
// never reach the rest of the roster.
(next, true)
+ } else if roster_walk
Review Comment:
Warning: Dropping the guard here means line 1283 re-acquires `routing_lock`
on the next turn, so the lock is held for roughly half the retry window. Keep
the guard across the retry instead.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -3137,9 +3674,33 @@ 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 {
Review Comment:
Warning: `promoted` advances only on success and every denial path uses
`continue`, so one call can drain all 64 queued requests with an awaited reply
send each. Break after a few consecutive denials.
##########
core/partitions/src/consumer_offset_capacity.rs:
##########
@@ -0,0 +1,679 @@
+// 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::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 with_entries<T>(
+ &self,
+ kind: ConsumerKind,
+ read: impl FnOnce(&HashMap<u32, DurableOffsetState>) -> T,
+ ) -> T {
+ read(&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);
+ let upper_bound = durable_count
+ .saturating_add(self.pending.borrow().len())
+ .saturating_add(self.provisional.borrow().len())
+ .saturating_add(self.stranded.borrow().len());
+ if !self.uncertain.get() && upper_bound < limit {
+ self.durable_warned.set(false);
+ return Ok(());
+ }
+ // 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) {
Review Comment:
Warning: `stranded` is cleared only by `clear_stranded`, while `rebuild` and
`mark_uncertain` leave it, so an unlinkable file latches permanent refusal of
new keys with no metric. Export stranded cardinality.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2959,6 +3195,59 @@ 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 entry in offset_dir_entries(dir) {
+ match entry {
+ OffsetDirEntry::Replacement(path) => {
+ if let Err(error) =
compio::fs::remove_file(&path).await {
+ tracing::warn!(
+ path,
+ %error,
+ "could not remove an abandoned offset
replacement"
+ );
+ }
+ }
+ OffsetDirEntry::Offset { id, path } => {
+ // Same policy as install: an unlinkable file strands
its
+ // key instead of failing the converge and looping.
+ match retry_offset_mutation(||
delete_persisted_offset(&path)).await {
+ Ok(_) =>
self.consumer_offset_capacity_for(kind).clear_stranded(id),
+ Err(error) => {
+
self.consumer_offset_capacity_for(kind).record_stranded(id);
+ tracing::warn!(
+ path,
+ consumer_id = id,
+ %error,
+ "converge could not remove a consumer
offset file"
+ );
+ }
+ }
+ }
+ }
+ }
+ if std::path::Path::new(dir).exists() {
Review Comment:
Warning: `Path::new(dir).exists()` is a blocking stat on the pump. Use the
`NotFound` tolerance already used in `flush_consumer_offset_directories`, which
is why `create_parent_dir` avoids the probe.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -1606,6 +1792,7 @@ where
if self.repair.is_some() {
return Err(PartitionTransferUnavailable::RepairInProgress);
}
+ self.validate_consumer_offset_transfer_counts()?;
Review Comment:
Simplification: `validate_consumer_offset_transfer_counts` runs again as the
first statement of `offsets_wire_snapshot`. Drop this call.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -4075,6 +4675,12 @@ where
let (frozen_batches, index_bytes, flush_index, batch_count,
committed_info, chunk_len) = {
let segment = self.log.active_segment();
let mut file_position = segment.size.as_bytes_u64();
+ let persisted_end = if file_position == 0 {
Review Comment:
Nit: The `persisted_end` fix is segment-flush idempotence for a repaired
batch, unrelated to consumer offset quotas. Split it out so it can be reverted
independently.
##########
core/partitions/src/consumer_offset_capacity.rs:
##########
@@ -0,0 +1,679 @@
+// 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::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 with_entries<T>(
+ &self,
+ kind: ConsumerKind,
+ read: impl FnOnce(&HashMap<u32, DurableOffsetState>) -> T,
+ ) -> T {
+ read(&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);
+ let upper_bound = durable_count
+ .saturating_add(self.pending.borrow().len())
+ .saturating_add(self.provisional.borrow().len())
Review Comment:
Warning: `provisional.len()` counts inactive tokens, so once the token map
reaches `limit` the fast path never fires and every `check` falls through to
`occupied`, allocating a `HashSet` per call. Count only active tokens.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -4367,7 +4979,26 @@ where
});
return;
}
+ }
+
+ // Commit replies and the applied frontier must follow directory
+ // durability. One sync per touched kind covers the whole walk. A sync
+ // failure is attributed to the batch boundary because one directory
+ // sync covers every delete in that kind, not one uniquely failing op.
+ if let Err(error) = self.flush_consumer_offset_directories().await {
+ if let Some(entry) = drained.last() {
Review Comment:
Nit: With an empty `drained` the sync error is dropped with no log and no
fence. Log unconditionally and keep only the fence inside the `if let`.
##########
core/partitions/src/offset_storage.rs:
##########
@@ -48,7 +51,7 @@ pub enum OffsetRecord {
/// A usable offset. `checksummed` is false for a bare offset predating the
/// checksum, read as-is and upgraded by the next write.
Value { offset: u64, checksummed: bool },
- /// Shorter than the value: a crash between `persist_offset`'s truncate
and write.
+ /// Shorter than the value, such as a legacy interrupted in-place write.
Review Comment:
Nit: In-place writes are not legacy. `write_in_place` is the default path
whenever `consumer_offset_enforce_fsync` is false, which is the shipped default.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2280,6 +2498,16 @@ where
}
}
+ let installed_end = staged.last().map(|meta| meta.end_offset);
+ let next_offset = offsets_wire
+ .next_offset
+ .max(installed_end.map_or(0, |end| end + 1));
+ let planned_offsets = self.plan_transfer_offset_writes(&offsets_wire,
next_offset)?;
+ // Write and data-sync every small offset record before the destructive
+ // segment swap. Only ignored, nonnumeric replacement siblings exist at
+ // this point, so a write fault leaves the old partition serviceable.
+ stage_offset_writes(&planned_offsets).await?;
Review Comment:
Nit: `stage_offset_writes` writes and syncs `.tmp` files here, below a
marker that says nothing may mutate. Reword the marker to say nothing may
mutate live state.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2341,7 +2627,15 @@ where
}
};
- if found {
+ // The live maps are what `get_consumer_offset` answers from, so a key
Review Comment:
Nit: The `replica_count() > 1` gate is inert, since every path that records
durable also inserts into the live map. Drop it so a single replica cannot
strand a durable slot.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -1551,6 +1642,28 @@ 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;
+/// First retry delay of [`retry_offset_mutation`]; each further retry doubles
it.
+const OFFSET_IO_BACKOFF_BASE: std::time::Duration =
std::time::Duration::from_millis(10);
+
+async fn retry_offset_mutation<T, E: fmt::Debug, F: Future<Output = Result<T,
E>>>(
Review Comment:
Nit: The first two attempts log at `debug!`, but the final
`operation().await` returns its error unlogged. Log the last attempt too.
##########
core/partitions/src/consumer_offset_capacity.rs:
##########
@@ -0,0 +1,679 @@
+// 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::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 with_entries<T>(
+ &self,
+ kind: ConsumerKind,
+ read: impl FnOnce(&HashMap<u32, DurableOffsetState>) -> T,
+ ) -> T {
+ read(&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);
+ let upper_bound = durable_count
+ .saturating_add(self.pending.borrow().len())
+ .saturating_add(self.provisional.borrow().len())
+ .saturating_add(self.stranded.borrow().len());
+ if !self.uncertain.get() && upper_bound < limit {
+ self.durable_warned.set(false);
+ return Ok(());
+ }
+ // 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) {
Review Comment:
Nit: `set_pending_count` assigns while `release_reservation` decrements the
same counter, and both take `&self`. Document that the partition's `&mut self`
call sites serialize them.
##########
core/shard/src/lib.rs:
##########
@@ -2009,6 +2024,32 @@ where
})
}
+ /// Submit an auto-commit back to the partition-owning shard's pump.
+ ///
+ /// # Errors
+ /// Returns a refusal if the local inbox cannot accept the frame.
+ pub fn submit_auto_commit_offset(
+ &self,
+ request: Message<RoutedRequestHeader>,
+ reservation: partitions::AutoCommitReservation,
+ ) -> Result<(), PartitionSubmitRefused> {
+ let frame = ShardFrame::lifecycle(LifecycleFrame::AutoCommitSubmit {
+ request,
+ reservation,
+ });
+ let sender = self
+ .senders
+ .get(usize::from(self.id))
+ .ok_or(PartitionSubmitRefused)?;
+ sender.try_send(frame).map_err(|error| {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
Review Comment:
Nit: A poll-side auto-commit refusal now lands in the same partition
frame-drop series as genuinely dropped frames. Use a distinct reason label.
##########
core/partitions/src/poll_plan.rs:
##########
@@ -861,55 +888,146 @@ impl AutoCommitCtx {
/// newer explicit store; the maps are lock-free (`papaya`), so this is
/// sound off the pump task.
#[allow(clippy::cast_possible_truncation)]
- pub(crate) fn apply(&self, offset: u64) {
- match &self.target {
+ pub(crate) fn apply(
+ self,
+ offset: u64,
+ ) -> Result<AutoCommitApplied, ConsumerOffsetCapacityError> {
+ let (kind, consumer_id) = self.kind_and_id();
+ let create = |path: Option<&str>| {
+ ConsumerOffset::new(
+ kind,
+ consumer_id,
+ offset,
+ path.map_or_else(String::new, |path|
format!("{path}/{consumer_id}")),
+ )
+ };
+ let previous_offset = match &self.target {
AutoCommitTarget::Consumer {
offsets,
consumer_id,
create_path,
- } => {
- let consumer_id = *consumer_id;
- let map: &ConsumerOffsets = offsets;
- upsert_offset_max(map, consumer_id as usize, offset, || {
- create_path.as_deref().map_or_else(
- || {
- ConsumerOffset::new(
- ConsumerKind::Consumer,
- consumer_id,
- 0,
- String::new(),
- )
- },
- |path|
ConsumerOffset::default_for_consumer(consumer_id, path),
- )
- });
- }
+ } => apply_local_offset(
+ offsets,
+ *consumer_id as usize,
+ offset,
+ &self.capacity,
+ self.durable.count(kind) >= self.capacity.limit(),
+ || create(create_path.as_deref()),
+ )?,
AutoCommitTarget::ConsumerGroup {
offsets,
group_id,
create_path,
- } => {
- let group_id = *group_id;
- let key = ConsumerGroupId(group_id as usize);
- let map: &ConsumerGroupOffsets = offsets;
- upsert_offset_max(map, key, offset, || {
- create_path.as_deref().map_or_else(
- || {
- ConsumerOffset::new(
- ConsumerKind::ConsumerGroup,
- group_id,
- 0,
- String::new(),
- )
- },
- |path| ConsumerOffset::default_for_consumer_group(key,
path),
- )
- });
- }
+ } => apply_local_offset(
+ offsets,
+ ConsumerGroupId(*group_id as usize),
+ offset,
+ &self.capacity,
+ self.durable.count(kind) >= self.capacity.limit(),
+ || create(create_path.as_deref()),
+ )?,
+ };
+ Ok(AutoCommitApplied {
+ kind,
+ consumer_id,
+ offset,
+ previous_offset,
+ target: self.target,
+ capacity: self.capacity,
+ durable: self.durable,
+ last_polled: None,
+ })
+ }
+}
+
+impl AutoCommitApplied {
+ /// Record the group handoff frontier only after poll admission succeeds.
+ pub fn mark_served(&self) {
+ if let Some(last_polled) = &self.last_polled {
+ last_polled.record(self.offset);
+ }
+ }
+ /// Reserve a durable key before the synthetic store is submitted.
+ /// Returns `None` when committed state already covers this offset.
+ ///
+ /// # Errors
+ /// Returns a capacity error when this is a new durable key and the
+ /// partition's per-kind limit has been reached.
+ pub fn reserve_durable(
+ &self,
+ ) -> Result<Option<AutoCommitReservation>, ConsumerOffsetCapacityError> {
+ if self
+ .durable
+ .covers(self.kind, self.consumer_id, self.offset)
+ {
+ return Ok(None);
+ }
+ self.capacity
+ .reserve_provisional(self.consumer_id, &self.durable)
+ .map(Some)
+ }
+
+ pub(crate) fn belongs_to(&self, durable: &Rc<DurableConsumerOffsets>) ->
bool {
+ Rc::ptr_eq(&self.durable, durable)
+ }
+
+ /// Undo this poll's eager update after synchronous admission fails. The
Review Comment:
Nit: `rollback_local_offset` does a bare non-monotone `store(previous)`, so
this no-yield contract is load bearing. Move the rollback decision inside
`execute` so a future caller cannot break it.
##########
core/partitions/src/offset_storage.rs:
##########
@@ -106,9 +109,40 @@ pub fn decode_offset_record(bytes: &[u8]) -> OffsetRecord {
/// Overwrite a consumer-offset file with `offset` and a checksum over it.
///
+/// Without `enforce_fsync` the file is rewritten in place. With it, the record
+/// goes to a sibling inode, is data-synced and renamed over the prior file,
so a
+/// failed write leaves the prior cursor intact. The replacement is tied to the
+/// same knob as the sync: without the sync neither the write nor the rename is
+/// ordered against a crash, so the extra inode and rename buy nothing. The
+/// caller syncs the parent directory afterwards.
Review Comment:
Nit: The caller syncs the parent directory only on the `enforce_fsync`
branch; the in-place branch never marks the directory dirty. Split the sentence
per branch.
##########
core/partitions/src/poll_plan.rs:
##########
@@ -861,55 +888,146 @@ impl AutoCommitCtx {
/// newer explicit store; the maps are lock-free (`papaya`), so this is
/// sound off the pump task.
#[allow(clippy::cast_possible_truncation)]
- pub(crate) fn apply(&self, offset: u64) {
- match &self.target {
+ pub(crate) fn apply(
+ self,
+ offset: u64,
+ ) -> Result<AutoCommitApplied, ConsumerOffsetCapacityError> {
+ let (kind, consumer_id) = self.kind_and_id();
+ let create = |path: Option<&str>| {
+ ConsumerOffset::new(
+ kind,
+ consumer_id,
+ offset,
+ path.map_or_else(String::new, |path|
format!("{path}/{consumer_id}")),
+ )
+ };
+ let previous_offset = match &self.target {
AutoCommitTarget::Consumer {
offsets,
consumer_id,
create_path,
- } => {
- let consumer_id = *consumer_id;
- let map: &ConsumerOffsets = offsets;
- upsert_offset_max(map, consumer_id as usize, offset, || {
- create_path.as_deref().map_or_else(
- || {
- ConsumerOffset::new(
- ConsumerKind::Consumer,
- consumer_id,
- 0,
- String::new(),
- )
- },
- |path|
ConsumerOffset::default_for_consumer(consumer_id, path),
- )
- });
- }
+ } => apply_local_offset(
+ offsets,
+ *consumer_id as usize,
+ offset,
+ &self.capacity,
+ self.durable.count(kind) >= self.capacity.limit(),
+ || create(create_path.as_deref()),
+ )?,
AutoCommitTarget::ConsumerGroup {
offsets,
group_id,
create_path,
- } => {
- let group_id = *group_id;
- let key = ConsumerGroupId(group_id as usize);
- let map: &ConsumerGroupOffsets = offsets;
- upsert_offset_max(map, key, offset, || {
- create_path.as_deref().map_or_else(
- || {
- ConsumerOffset::new(
- ConsumerKind::ConsumerGroup,
- group_id,
- 0,
- String::new(),
- )
- },
- |path| ConsumerOffset::default_for_consumer_group(key,
path),
- )
- });
- }
+ } => apply_local_offset(
+ offsets,
+ ConsumerGroupId(*group_id as usize),
+ offset,
+ &self.capacity,
+ self.durable.count(kind) >= self.capacity.limit(),
+ || create(create_path.as_deref()),
+ )?,
+ };
+ Ok(AutoCommitApplied {
+ kind,
+ consumer_id,
+ offset,
+ previous_offset,
+ target: self.target,
+ capacity: self.capacity,
+ durable: self.durable,
+ last_polled: None,
+ })
+ }
+}
+
+impl AutoCommitApplied {
+ /// Record the group handoff frontier only after poll admission succeeds.
+ pub fn mark_served(&self) {
+ if let Some(last_polled) = &self.last_polled {
+ last_polled.record(self.offset);
+ }
+ }
+ /// Reserve a durable key before the synthetic store is submitted.
+ /// Returns `None` when committed state already covers this offset.
+ ///
+ /// # Errors
+ /// Returns a capacity error when this is a new durable key and the
+ /// partition's per-kind limit has been reached.
+ pub fn reserve_durable(
+ &self,
+ ) -> Result<Option<AutoCommitReservation>, ConsumerOffsetCapacityError> {
+ if self
+ .durable
+ .covers(self.kind, self.consumer_id, self.offset)
+ {
+ return Ok(None);
+ }
+ self.capacity
+ .reserve_provisional(self.consumer_id, &self.durable)
+ .map(Some)
+ }
+
+ pub(crate) fn belongs_to(&self, durable: &Rc<DurableConsumerOffsets>) ->
bool {
+ Rc::ptr_eq(&self.durable, durable)
+ }
+
+ /// Undo this poll's eager update after synchronous admission fails. The
+ /// caller must not yield between execution and this rollback.
+ pub fn rollback_created(&self) {
+ match &self.target {
+ AutoCommitTarget::Consumer {
+ offsets,
+ consumer_id,
+ ..
+ } => rollback_local_offset(offsets, *consumer_id as usize,
self.previous_offset),
+ AutoCommitTarget::ConsumerGroup {
+ offsets, group_id, ..
+ } => rollback_local_offset(
+ offsets,
+ ConsumerGroupId(*group_id as usize),
+ self.previous_offset,
+ ),
+ }
+ if self.previous_offset.is_none() {
+ self.capacity.note_local_key_change();
+ self.capacity.forget_inactive_provisional(self.consumer_id);
}
}
}
+fn rollback_local_offset<K: Hash + Eq + Send + Sync + Copy>(
+ map: &papaya::HashMap<K, ConsumerOffset>,
+ key: K,
+ previous: Option<u64>,
+) {
+ let guard = map.pin();
+ if let Some(previous) = previous {
+ if let Some(entry) = guard.get(&key) {
+ entry.offset.store(previous, Ordering::Relaxed);
+ }
+ } else {
+ guard.remove(&key);
+ }
+}
+
+fn apply_local_offset<K: Hash + Eq + Clone + Send + Sync>(
+ map: &papaya::HashMap<K, ConsumerOffset>,
+ key: K,
+ offset: u64,
+ capacity: &ConsumerOffsetCapacity,
+ durable_full: bool,
+ create: impl FnOnce() -> ConsumerOffset,
+) -> Result<Option<u64>, ConsumerOffsetCapacityError> {
+ let guard = map.pin();
+ if let Some(existing) = guard.get(&key) {
+ return Ok(Some(existing.offset.fetch_max(offset, Ordering::Relaxed)));
+ }
+ capacity.admit_local_map_key(guard.len(), durable_full)?;
Review Comment:
Nit: The `len()` read and the insert are not atomic on this lock-free map.
Note that the bound holds only because polls for a partition run on its own
shard thread.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -231,60 +257,96 @@ fn submit_auto_commit<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
namespace: IggyNamespace,
applied: &AutoCommitApplied,
-) where
+) -> Result<(), IggyError>
+where
B: ShellBus,
MJ: JournalHandle + 'static,
MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
S: 'static,
SB: SuperblockStore + 'static,
{
- enum AutoCommitGate {
- Submit,
- Covered,
- NotPrimary,
- }
- let gate = shard
+ let primary = shard
Review Comment:
Nit: `record_consumer_offset_denied` covers only the `reserve_durable` path,
so pump-side denials for auto-commit clients are never counted. Count them, or
state that the metric is best effort.
##########
core/bench/src/benchmarks/common.rs:
##########
@@ -44,7 +44,11 @@ pub async fn create_consumer(
"Consumer #{} → joining consumer group #{}",
consumer_id, consumer_group_id
);
- let cg_identifier =
Identifier::try_from(*consumer_group_id).unwrap();
+ // By name: the group was created by name and the server assigns
its
Review Comment:
Nit: The consumer group addressing fixes here and at line 216 are unrelated
to consumer offset quotas. Split them into their own commit.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -1361,9 +1370,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}...");
Review Comment:
Nit: Every poll error now costs a full `polling_retry_interval`, one second
by default, before the caller sees it. Use a shorter delay for errors the
caller handles itself.
##########
core/server/src/boot/recovery.rs:
##########
@@ -331,6 +332,14 @@ const _: () = assert!(
const _: () = assert!(
configs::partition::PARTITION_DEDUP_CLIENTS_DEFAULT ==
consensus::PARTITION_DEDUP_CLIENTS_MAX
);
+const _: () = assert!(
+ configs::partition::PARTITION_CONSUMER_OFFSETS_DEFAULT
+ == partitions::DEFAULT_CONSUMER_OFFSETS_MAX
+);
+const _: () = assert!(
+ 4 * configs::partition::PARTITION_CONSUMER_OFFSETS_CEILING
Review Comment:
Nit: `4 * PARTITION_CONSUMER_OFFSETS_CEILING` equals
`CONSUMER_OFFSETS_ENTRIES_MAX` exactly, so no headroom is left. Say so in the
message, or lower the ceiling.
##########
core/integration/tests/server/consumer_offset_quota_vsr.rs:
##########
@@ -0,0 +1,490 @@
+// 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 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 group_file_count =
integration::harness::disk::consumer_offset_file_ids(
+ &harness.server().data_path(),
+ stream_details.id,
+ topic_details.id,
+ PARTITION_ID,
+ ConsumerKind::ConsumerGroup,
+ )
+ .map_or(0, |ids| ids.len());
+ assert_eq!(group_file_count, 0);
+
+ let named_group = StoreConsumerOffsetRequest {
+ consumer:
WireConsumer::consumer_group(WireIdentifier::named("unknown-group").unwrap()),
+ 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,
+ 2,
+ named_group.len(),
+ );
+ let (reply, _) = raw_tcp::exchange(&mut raw, &header, &named_group).await;
+ assert_eq!(
+ raw_tcp::reply_status(&reply),
+ IggyError::ConsumerGroupNameNotFound("unknown-group".to_owned(),
topic.clone()).as_code()
+ );
+
+ let unknown_stream = StoreConsumerOffsetRequest {
+ consumer: WireConsumer::consumer_group(WireIdentifier::Numeric(999)),
+ stream_id: WireIdentifier::Numeric(999_999),
+ 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,
+ 3,
+ unknown_stream.len(),
+ );
+ let (reply, _) = raw_tcp::exchange(&mut raw, &header,
&unknown_stream).await;
+ assert_eq!(
+ raw_tcp::reply_status(&reply),
+ IggyError::ResourceNotFound(String::new()).as_code(),
+ "a missing stream is not reported as a missing group"
+ );
+
+ let http = HttpClient::login_root(harness).await;
+ let response = http
+ .client
+ .put(http.url(&format!(
+ "/streams/{STREAM_NAME}/topics/{TOPIC_NAME}/consumer-offsets"
+ )))
+ .bearer_auth(&http.token)
+ .json(&StoreConsumerOffset {
+ consumer: Consumer::new(Identifier::numeric(6).expect("consumer
identifier")),
+ partition_id: Some(PARTITION_ID),
+ offset: 0,
+ })
+ .send()
+ .await
+ .expect("HTTP capacity request");
+ assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+ let body: serde_json::Value = response.json().await.expect("HTTP error
body");
+ assert_eq!(body["id"], 3024);
+ assert_eq!(body["code"], "too_many_consumer_offsets");
+
+ let metrics = http
+ .client
+ .get(http.url("/metrics"))
+ .bearer_auth(&http.token)
+ .send()
+ .await
+ .expect("metrics response")
+ .text()
+ .await
+ .expect("metrics text");
+ let denied: u64 = metrics
Review Comment:
Nit: This sums every denied-total series and expects 3, so a denial added
anywhere else in the test breaks it confusingly. Assert the consumer-kind
series instead.
--
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]
