You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2026/09/05 06:27:13 UTC
(iggy) 02/02: fix(partitions): bound consumer offset state per partition
This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch offset_hardening
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 7bd65595843b59e15d957141743c1844dab89840
Author: spetz <pi...@gmail.com>
AuthorDate: Sat Sep 5 08:27:00 2026 +0200
fix(partitions): bound consumer offset state per partition
---
Cargo.lock | 22 +-
Cargo.toml | 10 +-
bdd/python/uv.lock | 2 +-
core/ai/mcp/Cargo.toml | 2 +-
core/bench/Cargo.toml | 2 +-
core/bench/dashboard/frontend/Cargo.toml | 2 +-
core/bench/dashboard/server/Cargo.toml | 2 +-
core/bench/report/Cargo.toml | 2 +-
core/binary_protocol/Cargo.toml | 2 +-
core/binary_protocol/src/primitives/ack_level.rs | 6 +-
.../consumer_offsets/delete_consumer_offset.rs | 4 +-
.../consumer_offsets/store_consumer_offset.rs | 4 +-
core/cli/Cargo.toml | 2 +-
core/common/Cargo.toml | 2 +-
core/common/src/error/iggy_error.rs | 2 +-
core/common/src/traits/message_client.rs | 4 +
core/connectors/sdk/Cargo.toml | 2 +-
.../tests/cluster/consumer_offset_quota.rs | 42 +-
.../tests/server/consumer_offset_quota_vsr.rs | 87 +++-
core/message_bus/src/lib.rs | 2 +-
core/metadata/src/impls/metadata.rs | 6 +-
core/metadata/src/stm/user.rs | 4 +-
core/partitions/src/consumer_offset_capacity.rs | 268 ++++++-------
core/partitions/src/iggy_partition.rs | 445 ++++++++++++++++++---
core/partitions/src/iggy_partitions.rs | 16 +
core/partitions/src/poll_plan.rs | 142 +++----
core/partitions/src/state_transfer.rs | 17 +-
core/sdk/Cargo.toml | 2 +-
core/sdk/src/clients/consumer.rs | 5 +
core/server/Cargo.toml | 2 +-
core/server/config.toml | 3 +
core/server/src/boot/recovery.rs | 2 +-
core/server/src/consumer_group.rs | 14 +-
core/server/src/dispatch/partition.rs | 47 +--
core/server/src/partition_helpers.rs | 10 +-
core/server/src/partition_reconciler.rs | 154 +++++--
core/server/src/responses.rs | 32 +-
core/shard/src/lib.rs | 43 +-
core/shard/src/metrics.rs | 21 +-
core/shard/src/router.rs | 9 +
core/simulator/src/lib.rs | 17 +-
core/simulator/src/workload/effect.rs | 2 +-
core/simulator/src/workload/shadow.rs | 2 +-
examples/node/package-lock.json | 2 +-
examples/python/uv.lock | 2 +-
foreign/go/contracts/version.go | 2 +-
foreign/go/errors/errors.yaml | 2 +-
foreign/go/errors/errors_gen.go | 2 +-
.../org/apache/iggy/exception/IggyErrorCode.java | 2 +
foreign/node/package-lock.json | 4 +-
foreign/node/package.json | 2 +-
foreign/node/src/wire/error.code.test.ts | 2 +-
foreign/node/src/wire/error.code.ts | 2 +-
foreign/python/Cargo.toml | 4 +-
foreign/python/pyproject.toml | 2 +-
foreign/python/uv.lock | 2 +-
56 files changed, 1021 insertions(+), 475 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 56ecb71dd..7856dd413 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1862,7 +1862,7 @@ checksum = "3a8241f3ebb85c056b509d4327ad0358fbbba6ffb340bf388f26350aeda225b1"
[[package]]
name = "bench-dashboard-frontend"
-version = "0.8.0-edge.3"
+version = "0.8.0-edge.4"
dependencies = [
"bench-dashboard-shared",
"bench-report",
@@ -1892,7 +1892,7 @@ dependencies = [
[[package]]
name = "bench-report"
-version = "0.4.0-edge.3"
+version = "0.4.0-edge.4"
dependencies = [
"charming",
"colored",
@@ -6594,7 +6594,7 @@ checksum = "cd62e6b5e86ea8eeeb8db1de02880a6abc01a397b2ebb64b5d74ac255318f5cb"
[[package]]
name = "iggy"
-version = "0.11.0-edge.6"
+version = "0.11.0-edge.7"
dependencies = [
"async-broadcast",
"async-dropper",
@@ -6628,7 +6628,7 @@ dependencies = [
[[package]]
name = "iggy-bench"
-version = "0.6.0-edge.6"
+version = "0.6.0-edge.7"
dependencies = [
"async-trait",
"bench-report",
@@ -6657,7 +6657,7 @@ dependencies = [
[[package]]
name = "iggy-bench-dashboard-server"
-version = "0.8.0-edge.3"
+version = "0.8.0-edge.4"
dependencies = [
"actix-cors",
"actix-files",
@@ -6685,7 +6685,7 @@ dependencies = [
[[package]]
name = "iggy-cli"
-version = "0.14.0-edge.6"
+version = "0.14.0-edge.7"
dependencies = [
"anyhow",
"apple-native-keyring-store",
@@ -6791,7 +6791,7 @@ dependencies = [
[[package]]
name = "iggy-mcp"
-version = "0.5.0-edge.5"
+version = "0.5.0-edge.6"
dependencies = [
"axum",
"axum-server",
@@ -6825,7 +6825,7 @@ dependencies = [
[[package]]
name = "iggy_binary_protocol"
-version = "0.11.0-edge.6"
+version = "0.11.0-edge.7"
dependencies = [
"aligned-vec",
"bytemuck",
@@ -6838,7 +6838,7 @@ dependencies = [
[[package]]
name = "iggy_common"
-version = "0.11.0-edge.6"
+version = "0.11.0-edge.7"
dependencies = [
"aes-gcm 0.11.1",
"async-broadcast",
@@ -7203,7 +7203,7 @@ dependencies = [
[[package]]
name = "iggy_connector_sdk"
-version = "0.4.0-edge.3"
+version = "0.4.0-edge.4"
dependencies = [
"anyhow",
"apache-avro 0.22.0",
@@ -12160,7 +12160,7 @@ dependencies = [
[[package]]
name = "server"
-version = "0.9.0-edge.6"
+version = "0.9.0-edge.7"
dependencies = [
"ahash 0.8.12",
"argon2",
diff --git a/Cargo.toml b/Cargo.toml
index af7a697ca..beb92e80f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -198,13 +198,13 @@ hyper-util = { version = "0.1.20", features = ["server-auto", "service"] }
iceberg = "0.10.1"
iceberg-catalog-rest = "0.10.1"
iceberg-storage-opendal = "0.10.1"
-iggy = { path = "core/sdk", version = "0.11.0-edge.6" }
-iggy-cli = { path = "core/cli", version = "0.14.0-edge.6" }
+iggy = { path = "core/sdk", version = "0.11.0-edge.7" }
+iggy-cli = { path = "core/cli", version = "0.14.0-edge.7" }
iggy-gateway-kafka = { path = "gateways/kafka" }
-iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.6" }
-iggy_common = { path = "core/common", version = "0.11.0-edge.6" }
+iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.7" }
+iggy_common = { path = "core/common", version = "0.11.0-edge.7" }
iggy_connector_doris_sink = { path = "core/connectors/sinks/doris_sink" }
-iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.4.0-edge.3" }
+iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.4.0-edge.4" }
indexmap = "2.14.1"
integration = { path = "core/integration" }
ipnet = "2.12.1"
diff --git a/bdd/python/uv.lock b/bdd/python/uv.lock
index 44c64ab55..b10e81b84 100644
--- a/bdd/python/uv.lock
+++ b/bdd/python/uv.lock
@@ -8,7 +8,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev6"
+version = "0.9.0.dev7"
source = { directory = "../../foreign/python" }
[package.metadata]
diff --git a/core/ai/mcp/Cargo.toml b/core/ai/mcp/Cargo.toml
index 5386c5c4a..1d1953f9f 100644
--- a/core/ai/mcp/Cargo.toml
+++ b/core/ai/mcp/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-mcp"
-version = "0.5.0-edge.5"
+version = "0.5.0-edge.6"
description = "MCP Server for Iggy message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/bench/Cargo.toml b/core/bench/Cargo.toml
index e2a3c8d8d..7a3f1ce42 100644
--- a/core/bench/Cargo.toml
+++ b/core/bench/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-bench"
-version = "0.6.0-edge.6"
+version = "0.6.0-edge.7"
edition = "2024"
license = "Apache-2.0"
repository = "https://github.com/apache/iggy"
diff --git a/core/bench/dashboard/frontend/Cargo.toml b/core/bench/dashboard/frontend/Cargo.toml
index 595cec1d0..6b023d631 100644
--- a/core/bench/dashboard/frontend/Cargo.toml
+++ b/core/bench/dashboard/frontend/Cargo.toml
@@ -18,7 +18,7 @@
[package]
name = "bench-dashboard-frontend"
license = "Apache-2.0"
-version = "0.8.0-edge.3"
+version = "0.8.0-edge.4"
edition = "2024"
publish = false
diff --git a/core/bench/dashboard/server/Cargo.toml b/core/bench/dashboard/server/Cargo.toml
index 726df33be..68570b112 100644
--- a/core/bench/dashboard/server/Cargo.toml
+++ b/core/bench/dashboard/server/Cargo.toml
@@ -18,7 +18,7 @@
[package]
name = "iggy-bench-dashboard-server"
license = "Apache-2.0"
-version = "0.8.0-edge.3"
+version = "0.8.0-edge.4"
edition = "2024"
publish = false
diff --git a/core/bench/report/Cargo.toml b/core/bench/report/Cargo.toml
index a418cdd25..bc5649a57 100644
--- a/core/bench/report/Cargo.toml
+++ b/core/bench/report/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "bench-report"
-version = "0.4.0-edge.3"
+version = "0.4.0-edge.4"
edition = "2024"
description = "Benchmark report and chart generation library for iggy-bench binary and iggy-benchmarks-dashboard web app"
license = "Apache-2.0"
diff --git a/core/binary_protocol/Cargo.toml b/core/binary_protocol/Cargo.toml
index 0a83a4c01..be97141e3 100644
--- a/core/binary_protocol/Cargo.toml
+++ b/core/binary_protocol/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_binary_protocol"
-version = "0.11.0-edge.6"
+version = "0.11.0-edge.7"
description = "Wire protocol types and codec for the Iggy binary protocol. Shared between server and SDK."
edition = "2024"
rust-version.workspace = true
diff --git a/core/binary_protocol/src/primitives/ack_level.rs b/core/binary_protocol/src/primitives/ack_level.rs
index a2ddb2389..a750b57d9 100644
--- a/core/binary_protocol/src/primitives/ack_level.rs
+++ b/core/binary_protocol/src/primitives/ack_level.rs
@@ -20,9 +20,9 @@ use crate::WireError;
/// Acknowledgement policy for consumer-offset write commands.
///
/// Wire format: single `u8` discriminant.
-/// - `NoAck(0)`: leader-local write only; respond as soon as the in-memory
-/// and on-disk state have been updated. Matches the fast path used by
-/// `PollMessages` auto-commit.
+/// - `NoAck(0)`: local fast path for a single-replica partition. Replicated
+/// partitions commit offset writes through VSR before replying, including
+/// when this acknowledgement value is selected.
/// - `Quorum(1)`: submit through the partition VSR consensus pipeline and
/// respond only after the write has been committed by a quorum of replicas.
/// This is the default for explicit client writes.
diff --git a/core/binary_protocol/src/requests/consumer_offsets/delete_consumer_offset.rs b/core/binary_protocol/src/requests/consumer_offsets/delete_consumer_offset.rs
index 014081794..e5ca29d1a 100644
--- a/core/binary_protocol/src/requests/consumer_offsets/delete_consumer_offset.rs
+++ b/core/binary_protocol/src/requests/consumer_offsets/delete_consumer_offset.rs
@@ -24,8 +24,8 @@ use bytes::{BufMut, BytesMut};
/// `DeleteConsumerOffset` request.
///
-/// Adds an `ack` byte: `NoAck` = leader-local fast path, `Quorum` = VSR
-/// pipeline.
+/// The `ack` byte selects the local fast path only for `NoAck` on a
+/// single-replica partition. Replicated partitions use VSR for both values.
///
/// Wire format:
/// ```text
diff --git a/core/binary_protocol/src/requests/consumer_offsets/store_consumer_offset.rs b/core/binary_protocol/src/requests/consumer_offsets/store_consumer_offset.rs
index 5261d5c53..228360798 100644
--- a/core/binary_protocol/src/requests/consumer_offsets/store_consumer_offset.rs
+++ b/core/binary_protocol/src/requests/consumer_offsets/store_consumer_offset.rs
@@ -24,8 +24,8 @@ use bytes::{BufMut, BytesMut};
/// `StoreConsumerOffset` request.
///
-/// Adds an `ack` byte: `NoAck` = leader-local fast path, `Quorum` = VSR
-/// pipeline.
+/// The `ack` byte selects the local fast path only for `NoAck` on a
+/// single-replica partition. Replicated partitions use VSR for both values.
///
/// Wire format:
/// ```text
diff --git a/core/cli/Cargo.toml b/core/cli/Cargo.toml
index a2f8756bb..f8a2b89f1 100644
--- a/core/cli/Cargo.toml
+++ b/core/cli/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-cli"
-version = "0.14.0-edge.6"
+version = "0.14.0-edge.7"
edition = "2024"
rust-version.workspace = true
authors = ["bartosz.ciesla@gmail.com"]
diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml
index 16aaa621c..2e059fc3e 100644
--- a/core/common/Cargo.toml
+++ b/core/common/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_common"
-version = "0.11.0-edge.6"
+version = "0.11.0-edge.7"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
rust-version.workspace = true
diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index b312d61dc..e77abebc5 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -333,7 +333,7 @@ pub enum IggyError {
NotResolvedConsumer(Identifier) = 3022,
#[error("Cannot open consumer offsets file for path: {0}")]
CannotOpenConsumerOffsetsFile(String) = 3023,
- #[error("Too many consumer offsets")]
+ #[error("Too many consumer offsets for partition")]
TooManyConsumerOffsets = 3024,
#[error("Segment not found")]
SegmentNotFound = 4000,
diff --git a/core/common/src/traits/message_client.rs b/core/common/src/traits/message_client.rs
index 18c6d2294..7e1dcd026 100644
--- a/core/common/src/traits/message_client.rs
+++ b/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
+ /// rejected with `TooManyConsumerOffsets` at the partition's configured
+ /// limit. That poll returns no messages. Existing keys remain writable,
+ /// and polling without auto-commit does not allocate a stored offset.
async fn poll_messages(
&self,
stream_id: &Identifier,
diff --git a/core/connectors/sdk/Cargo.toml b/core/connectors/sdk/Cargo.toml
index fae1b4a0a..5d6b98daa 100644
--- a/core/connectors/sdk/Cargo.toml
+++ b/core/connectors/sdk/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_sdk"
-version = "0.4.0-edge.3"
+version = "0.4.0-edge.4"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/integration/tests/cluster/consumer_offset_quota.rs b/core/integration/tests/cluster/consumer_offset_quota.rs
index c39bab514..3aa9a7d9d 100644
--- a/core/integration/tests/cluster/consumer_offset_quota.rs
+++ b/core/integration/tests/cluster/consumer_offset_quota.rs
@@ -141,6 +141,22 @@ async fn given_replicated_partition_when_no_ack_offsets_mutate_should_converge_a
)
.await;
+ let auto_commit_consumer = Consumer::new(Identifier::numeric(1).expect("consumer identifier"));
+ let polled = client
+ .poll_messages(
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ &auto_commit_consumer,
+ &PollingStrategy::first(),
+ 1,
+ true,
+ )
+ .await
+ .expect("auto-commit poll within limit");
+ assert_eq!(polled.messages.len(), 1);
+ wait_for_file_state(harness, stream_details.id, topic_details.id, 1, true).await;
+
for consumer_id in 1..=4 {
client
.store_consumer_offset(
@@ -306,23 +322,15 @@ async fn wait_for_max_file_count(harness: &TestHarness, stream_id: u32, topic_id
loop {
let counts: Vec<usize> = (0..harness.cluster_size())
.map(|node| {
- let dir = offset_file(harness, node, stream_id, topic_id, 0)
- .parent()
- .expect("offset file has parent")
- .to_path_buf();
- std::fs::read_dir(dir)
- .map(|entries| {
- entries
- .filter_map(Result::ok)
- .filter(|entry| {
- entry
- .file_name()
- .to_str()
- .is_some_and(|name| name.parse::<u32>().is_ok())
- })
- .count()
- })
- .unwrap_or_default()
+ disk::consumer_offset_file_ids(
+ &harness.node(node).data_path(),
+ stream_id,
+ topic_id,
+ PARTITION_ID,
+ ConsumerKind::Consumer,
+ )
+ .expect("consumer offset directory")
+ .len()
})
.collect();
if counts.iter().all(|count| *count == max) {
diff --git a/core/integration/tests/server/consumer_offset_quota_vsr.rs b/core/integration/tests/server/consumer_offset_quota_vsr.rs
index 3849ae096..2ccac51c4 100644
--- a/core/integration/tests/server/consumer_offset_quota_vsr.rs
+++ b/core/integration/tests/server/consumer_offset_quota_vsr.rs
@@ -107,6 +107,49 @@ async fn given_full_consumer_offset_table_when_creating_another_should_reject_wi
.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(
@@ -205,23 +248,23 @@ async fn given_full_consumer_offset_table_when_creating_another_should_reject_wi
let (reply, _) = raw_tcp::exchange(&mut raw, &header, &unresolved_group).await;
assert_eq!(
raw_tcp::reply_status(&reply),
- IggyError::InvalidIdentifier.as_code()
+ 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 = fs::read_dir(&offsets_dir)
- .expect("consumer offsets directory")
- .filter_map(Result::ok)
- .filter(|entry| {
- entry
- .file_name()
- .to_str()
- .is_some_and(|name| name.parse::<u32>().is_ok())
- })
- .count();
+ 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
.parent()
@@ -232,6 +275,28 @@ async fn given_full_consumer_offset_table_when_creating_another_should_reject_wi
.unwrap_or_default();
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 http = HttpClient::login_root(harness).await;
let response = http
.client
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index 74836bb8e..d8038b612 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -797,7 +797,7 @@ impl IggyMessageBus {
}
/// Construct a bus with explicit runtime tunables and a pre-allocated
- /// owner table. Server-ng bootstrap uses this so every shard's bus
+ /// owner table. Server bootstrap uses this so every shard's bus
/// shares the same atomic slots; tests use [`Self::with_tunables`]
/// which allocates a fresh table per bus.
///
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index 726a013c1..5358fd0d4 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -679,7 +679,7 @@ pub fn apply_committed_prepare<M>(
pub type CommitNotifier = std::rc::Rc<dyn Fn(Operation)>;
pub struct IggyMetadata<C, J, S, M, SB = PingPongSuperblock> {
- /// `Some` on shard 0, `None` on other shards. Server-ng bootstrap
+ /// `Some` on shard 0, `None` on other shards. Server bootstrap
/// holds the invariant: only shard 0 owns the metadata consensus
/// replica; every other shard reconstructs `mux_stm` from the
/// `MetadataHandoff::Waiter` factory bundle broadcast by shard 0
@@ -897,7 +897,7 @@ impl<C, J, S, M, SB> IggyMetadata<C, J, S, M, SB> {
}
/// Install (or replace) the post-commit notifier. Passing `None`
- /// removes any previous one. Server-ng bootstrap calls this on shard 0
+ /// removes any previous one. Server bootstrap calls this on shard 0
/// only; peer shards never commit metadata locally.
pub fn set_commit_notifier(&self, notifier: Option<CommitNotifier>) {
*self.commit_notifier.borrow_mut() = notifier;
@@ -906,7 +906,7 @@ impl<C, J, S, M, SB> IggyMetadata<C, J, S, M, SB> {
/// Seed the coordinator's last-checkpoint pairing at boot from the recovered
/// snapshot, so the first post-boot view-change superblock write records the real
/// `(checkpoint_op, checksum)` instead of `(0, 0)`. No-op without a coordinator
- /// (peer shards, the simulator). Server-ng bootstrap calls this on shard 0 after
+ /// (peer shards, the simulator). Server bootstrap calls this on shard 0 after
/// cross-checking the pairing.
pub fn seed_checkpoint_ref(&self, checkpoint_op: u64, checkpoint_checksum: u128) {
if let Some(coordinator) = &self.coordinator {
diff --git a/core/metadata/src/stm/user.rs b/core/metadata/src/stm/user.rs
index 1e79d7853..c78ea7507 100644
--- a/core/metadata/src/stm/user.rs
+++ b/core/metadata/src/stm/user.rs
@@ -642,14 +642,14 @@ impl StateHandler for UpdatePermissionsRequest {
/// The success reply here is deliberately empty: the raw token the caller needs
/// is the one thing this apply must never see.
///
-/// The primary mints the raw token and its hash at ingress (server-ng
+/// The primary mints the raw token and its hash at ingress (server
/// `pat::rewrite_pat_request_for_user`) and replicates only the hash. Minting
/// inside this apply would call `ring::rand` on every replica and diverge the
/// token index, and replicating the raw token would persist a live credential in
/// every WAL and snapshot. So the raw token leaves the primary by a side channel
/// (`maybe_rewrite_pat_request` returns it alongside the rewritten request) and
/// the home shard splices it into this op's reply as a typed
-/// `RawPersonalAccessTokenResponse` (server-ng `responses::build_raw_pat_reply`).
+/// `RawPersonalAccessTokenResponse` (server `responses::build_raw_pat_reply`).
///
/// One consequence rides on that: the secret exists only on the wire of the
/// original reply, so a replayed request cannot be served from the client-table
diff --git a/core/partitions/src/consumer_offset_capacity.rs b/core/partitions/src/consumer_offset_capacity.rs
index 6523b2614..27c786ceb 100644
--- a/core/partitions/src/consumer_offset_capacity.rs
+++ b/core/partitions/src/consumer_offset_capacity.rs
@@ -19,6 +19,7 @@ use iggy_common::ConsumerKind;
use std::cell::{Cell, RefCell};
use std::collections::{HashMap, HashSet};
use std::rc::Rc;
+use std::sync::{Arc, Weak};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DurableOffsetState {
@@ -28,25 +29,30 @@ pub struct DurableOffsetState {
#[derive(Debug, Default)]
pub struct DurableConsumerOffsets {
- entries: RefCell<HashMap<(ConsumerKind, u32), DurableOffsetState>>,
- consumer_count: Cell<usize>,
- group_count: Cell<usize>,
+ consumers: RefCell<HashMap<u32, DurableOffsetState>>,
+ groups: RefCell<HashMap<u32, DurableOffsetState>>,
}
impl DurableConsumerOffsets {
pub(crate) fn get(&self, kind: ConsumerKind, id: u32) -> Option<DurableOffsetState> {
- self.entries.borrow().get(&(kind, id)).copied()
+ self.entries(kind).borrow().get(&id).copied()
}
pub(crate) fn contains(&self, kind: ConsumerKind, id: u32) -> bool {
- self.entries.borrow().contains_key(&(kind, id))
+ self.entries(kind).borrow().contains_key(&id)
}
- pub(crate) const fn count(&self, kind: ConsumerKind) -> usize {
- match kind {
- ConsumerKind::Consumer => self.consumer_count.get(),
- ConsumerKind::ConsumerGroup => self.group_count.get(),
- }
+ 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(
@@ -56,8 +62,7 @@ impl DurableConsumerOffsets {
committed_offset: u64,
persisted_high_water: Option<u64>,
) {
- self.insert_or_update(
- kind,
+ self.entries(kind).borrow_mut().insert(
id,
DurableOffsetState {
committed_offset,
@@ -71,99 +76,70 @@ impl DurableConsumerOffsets {
kind: ConsumerKind,
id: u32,
committed_offset: u64,
- persisted_high_water: Option<u64>,
+ persisted_high_water: u64,
) {
- let mut entries = self.entries.borrow_mut();
- if let Some(state) = entries.get_mut(&(kind, id)) {
- state.committed_offset = state.committed_offset.max(committed_offset);
- if let Some(high_water) = persisted_high_water {
- state.persisted_high_water = Some(
- state
- .persisted_high_water
- .map_or(high_water, |current| current.max(high_water)),
- );
- }
- return;
- }
- entries.insert(
- (kind, id),
- DurableOffsetState {
- committed_offset,
- persisted_high_water,
- },
+ 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),
);
- drop(entries);
- self.increment(kind);
}
pub(crate) fn mark_persisted(&self, kind: ConsumerKind, id: u32, high_water: u64) {
- if let Some(state) = self.entries.borrow_mut().get_mut(&(kind, id)) {
+ 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 {
- if self.entries.borrow_mut().remove(&(kind, id)).is_none() {
- return false;
- }
- self.decrement(kind);
- true
+ self.entries(kind).borrow_mut().remove(&id).is_some()
}
pub(crate) fn clear(&self) {
- self.entries.borrow_mut().clear();
- self.consumer_count.set(0);
- self.group_count.set(0);
+ self.consumers.borrow_mut().clear();
+ self.groups.borrow_mut().clear();
}
pub(crate) fn committed_entries(&self, kind: ConsumerKind) -> Vec<(u32, u64)> {
- self.entries
+ self.entries(kind)
.borrow()
.iter()
- .filter_map(|((entry_kind, id), state)| {
- (*entry_kind == kind).then_some((*id, state.committed_offset))
- })
+ .map(|(id, state)| (*id, state.committed_offset))
.collect()
}
- fn insert_or_update(&self, kind: ConsumerKind, id: u32, state: DurableOffsetState) {
- let inserted = self
- .entries
- .borrow_mut()
- .insert((kind, id), state)
- .is_none();
- if inserted {
- self.increment(kind);
- }
- }
-
- fn increment(&self, kind: ConsumerKind) {
- match kind {
- ConsumerKind::Consumer => self.consumer_count.set(self.consumer_count.get() + 1),
- ConsumerKind::ConsumerGroup => self.group_count.set(self.group_count.get() + 1),
- }
- }
-
- fn decrement(&self, kind: ConsumerKind) {
+ const fn entries(&self, kind: ConsumerKind) -> &RefCell<HashMap<u32, DurableOffsetState>> {
match kind {
- ConsumerKind::Consumer => self.consumer_count.set(self.consumer_count.get() - 1),
- ConsumerKind::ConsumerGroup => self.group_count.set(self.group_count.get() - 1),
+ ConsumerKind::Consumer => &self.consumers,
+ ConsumerKind::ConsumerGroup => &self.groups,
}
}
}
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum CapacityReservation {
- Existing,
- Reserved,
-}
-
#[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)]
@@ -171,7 +147,7 @@ pub struct ConsumerOffsetCapacity {
kind: ConsumerKind,
limit: Cell<usize>,
pending: RefCell<HashMap<u32, usize>>,
- provisional: RefCell<HashMap<u32, usize>>,
+ provisional: RefCell<HashMap<u32, Weak<()>>>,
stranded: RefCell<HashSet<u32>>,
uncertain: Cell<bool>,
durable_warned: Cell<bool>,
@@ -196,43 +172,52 @@ impl ConsumerOffsetCapacity {
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<CapacityReservation, ConsumerOffsetCapacityError> {
- let existing = self.check(id, durable)?;
+ ) -> Result<(), ConsumerOffsetCapacityError> {
+ self.check(id, durable)?;
*self.pending.borrow_mut().entry(id).or_default() += 1;
- Ok(if existing {
- CapacityReservation::Existing
- } else {
- CapacityReservation::Reserved
- })
+ Ok(())
}
pub(crate) fn check(
&self,
id: u32,
durable: &DurableConsumerOffsets,
- ) -> Result<bool, ConsumerOffsetCapacityError> {
+ ) -> Result<(), ConsumerOffsetCapacityError> {
+ self.rearm_if_below_limit(durable);
if durable.contains(self.kind, id)
|| self.pending.borrow().contains_key(&id)
- || self.provisional.borrow().contains_key(&id)
+ || self
+ .provisional
+ .borrow()
+ .get(&id)
+ .is_some_and(|token| token.strong_count() > 0)
|| self.stranded.borrow().contains(&id)
{
- return Ok(true);
+ return Ok(());
}
+ self.provisional
+ .borrow_mut()
+ .retain(|_, token| token.strong_count() > 0);
let occupied = self.occupied(durable);
let limit = self.limit.get();
if self.uncertain.get() || occupied >= limit {
return Err(ConsumerOffsetCapacityError {
kind: self.kind,
- occupied: occupied.max(limit),
+ occupied,
limit,
first_in_episode: !self.durable_warned.replace(true),
+ uncertain: self.uncertain.get(),
});
}
- Ok(false)
+ Ok(())
}
pub(crate) fn reserve_provisional(
@@ -241,14 +226,41 @@ impl ConsumerOffsetCapacity {
durable: &Rc<DurableConsumerOffsets>,
) -> Result<AutoCommitReservation, ConsumerOffsetCapacityError> {
self.check(id, durable)?;
- *self.provisional.borrow_mut().entry(id).or_default() += 1;
+ let mut provisional = self.provisional.borrow_mut();
+ let token = provisional
+ .get(&id)
+ .and_then(Weak::upgrade)
+ .unwrap_or_else(|| {
+ let token = Arc::new(());
+ provisional.insert(id, Arc::downgrade(&token));
+ token
+ });
Ok(AutoCommitReservation {
- capacity: Rc::clone(self),
- durable: Rc::clone(durable),
+ token,
+ kind: self.kind,
consumer_id: id,
})
}
+ pub(crate) fn owns(&self, reservation: &AutoCommitReservation) -> bool {
+ reservation.kind == self.kind
+ && self
+ .provisional
+ .borrow()
+ .get(&reservation.consumer_id)
+ .is_some_and(|token| std::ptr::eq(token.as_ptr(), Arc::as_ptr(&reservation.token)))
+ }
+
+ pub(crate) fn protects(&self, id: u32, durable: &DurableConsumerOffsets) -> bool {
+ durable.contains(self.kind, id)
+ || self.pending.borrow().contains_key(&id)
+ || self
+ .provisional
+ .borrow()
+ .get(&id)
+ .is_some_and(|token| token.strong_count() > 0)
+ }
+
pub(crate) fn set_pending_count(&self, id: u32, count: usize) {
if count == 0 {
self.pending.borrow_mut().remove(&id);
@@ -257,11 +269,6 @@ impl ConsumerOffsetCapacity {
}
}
- pub(crate) fn promote_to_durable(&self, id: u32) {
- self.release_reservation(id);
- self.stranded.borrow_mut().remove(&id);
- }
-
pub(crate) fn release_reservation(&self, id: u32) {
let mut pending = self.pending.borrow_mut();
let Some(count) = pending.get_mut(&id) else {
@@ -315,6 +322,9 @@ impl ConsumerOffsetCapacity {
}
pub(crate) fn rearm_if_below_limit(&self, durable: &DurableConsumerOffsets) {
+ if !self.durable_warned.get() || self.uncertain.get() {
+ return;
+ }
if self.occupied(durable) < self.limit.get() {
self.durable_warned.set(false);
}
@@ -333,6 +343,7 @@ impl ConsumerOffsetCapacity {
occupied: map_len,
limit,
first_in_episode: !self.map_warned.replace(true),
+ uncertain: false,
})
}
@@ -342,51 +353,36 @@ impl ConsumerOffsetCapacity {
}
}
- pub(crate) fn occupied_count(&self, durable: &DurableConsumerOffsets) -> usize {
- self.occupied(durable)
- }
-
- fn occupied(&self, durable: &DurableConsumerOffsets) -> usize {
+ pub(crate) fn occupied(&self, durable: &DurableConsumerOffsets) -> usize {
let pending = self.pending.borrow();
let provisional = self.provisional.borrow();
let stranded = self.stranded.borrow();
durable.count(self.kind)
+ pending
.keys()
- .chain(provisional.keys().filter(|id| !pending.contains_key(id)))
.chain(
- stranded
+ provisional
.iter()
- .filter(|id| !pending.contains_key(id) && !provisional.contains_key(id)),
+ .filter(|(id, token)| !pending.contains_key(id) && token.strong_count() > 0)
+ .map(|(id, _)| id),
)
+ .chain(stranded.iter().filter(|id| {
+ !pending.contains_key(id)
+ && provisional
+ .get(id)
+ .is_none_or(|token| token.strong_count() == 0)
+ }))
.filter(|id| !durable.contains(self.kind, **id))
.count()
}
}
-/// Owns a poll's reservation until its submitted operation either commits or
-/// loses its reply channel.
-///
-/// Journal reservations are accounted independently,
-/// so cancellation and view changes cannot release another operation's slot.
+/// Keeps a provisional key occupied until the pump admits or drops its request.
+#[derive(Debug)]
pub struct AutoCommitReservation {
- capacity: Rc<ConsumerOffsetCapacity>,
- durable: Rc<DurableConsumerOffsets>,
- consumer_id: u32,
-}
-
-impl Drop for AutoCommitReservation {
- fn drop(&mut self) {
- let mut provisional = self.capacity.provisional.borrow_mut();
- if let Some(count) = provisional.get_mut(&self.consumer_id) {
- *count -= 1;
- if *count == 0 {
- provisional.remove(&self.consumer_id);
- }
- }
- drop(provisional);
- self.capacity.rearm_if_below_limit(&self.durable);
- }
+ token: Arc<()>,
+ pub(crate) kind: ConsumerKind,
+ pub(crate) consumer_id: u32,
}
#[cfg(test)]
@@ -421,9 +417,9 @@ mod tests {
assert!(capacity.check(9, &durable).is_err());
drop(first);
assert!(capacity.check(9, &durable).is_ok());
- assert_eq!(capacity.occupied_count(&durable), 1);
+ assert_eq!(capacity.occupied(&durable), 1);
drop(second);
- assert_eq!(capacity.occupied_count(&durable), 0);
+ assert_eq!(capacity.occupied(&durable), 0);
}
#[test]
@@ -443,14 +439,8 @@ mod tests {
fn given_same_pending_key_when_reserved_twice_should_consume_one_slot() {
let durable = DurableConsumerOffsets::default();
let capacity = ConsumerOffsetCapacity::new(ConsumerKind::Consumer, 1);
- assert_eq!(
- capacity.try_reserve(7, &durable),
- Ok(CapacityReservation::Reserved)
- );
- assert_eq!(
- capacity.try_reserve(7, &durable),
- Ok(CapacityReservation::Existing)
- );
+ assert_eq!(capacity.try_reserve(7, &durable), Ok(()));
+ assert_eq!(capacity.try_reserve(7, &durable), Ok(()));
assert!(capacity.try_reserve(8, &durable).is_err());
capacity.release_reservation(7);
assert!(
@@ -484,7 +474,15 @@ mod tests {
let error = capacity
.try_reserve(7, &durable)
.expect_err("unknown pending state must block new keys");
- assert_eq!(error.occupied, 4);
+ assert_eq!(error.occupied, 0);
+ assert!(error.uncertain);
+ assert!(error.first_in_episode);
+ assert!(
+ !capacity
+ .try_reserve(8, &durable)
+ .expect_err("the same uncertain episode stays closed")
+ .first_in_episode
+ );
}
#[test]
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index b42d8cb01..352dc60ea 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -16,8 +16,7 @@
// under the License.
use crate::consumer_offset_capacity::{
- CapacityReservation, ConsumerOffsetCapacity, ConsumerOffsetCapacityError,
- DurableConsumerOffsets,
+ ConsumerOffsetCapacity, ConsumerOffsetCapacityError, DurableConsumerOffsets,
};
use crate::iggy_index_writer::IggyIndexWriter;
use crate::journal::{MessageLookup, PartitionJournal, PartitionJournalMemStorage};
@@ -209,6 +208,8 @@ where
/// server down without the partition moving again in the meantime.
fatal: Option<FatalCommit>,
pub(crate) pending_consumer_offset_commits: HashMap<u64, PendingConsumerOffsetCommit>,
+ pub(crate) queued_auto_commit_reservations:
+ RefCell<HashMap<(ConsumerKind, u32), Vec<crate::AutoCommitReservation>>>,
/// Committed consumer-offset membership and values. This is deliberately
/// separate from the eager poll maps because follower-local and uncommitted
/// auto-commit progress must never consume a durable slot or enter a state
@@ -556,6 +557,7 @@ where
installed_frontier: None,
fatal: None,
pending_consumer_offset_commits: HashMap::new(),
+ queued_auto_commit_reservations: RefCell::new(HashMap::new()),
durable_consumer_offsets: Rc::new(DurableConsumerOffsets::default()),
consumer_offset_capacity: Rc::new(ConsumerOffsetCapacity::new(
ConsumerKind::Consumer,
@@ -2009,11 +2011,13 @@ where
);
}
- pub fn seed_stranded_consumer_offset(&self, kind: ConsumerKind, consumer_id: u32) {
+ pub fn seed_stranded_consumer_offset(&self, kind: ConsumerKind, consumer_id: u32) -> bool {
if !self.durable_consumer_offsets.contains(kind, consumer_id) {
self.consumer_offset_capacity_for(kind)
.record_stranded(consumer_id);
+ return true;
}
+ false
}
#[must_use]
@@ -2024,7 +2028,7 @@ where
#[must_use]
pub fn occupied_consumer_offset_count(&self, kind: ConsumerKind) -> usize {
self.consumer_offset_capacity_for(kind)
- .occupied_count(&self.durable_consumer_offsets)
+ .occupied(&self.durable_consumer_offsets)
}
#[must_use]
@@ -2118,6 +2122,10 @@ where
) -> Result<(), IggyError> {
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
@@ -2134,35 +2142,37 @@ where
.durable_consumer_offsets
.get(pending.kind, pending.consumer_id);
let persisted_high_water = match (path.as_deref(), tracked) {
- (None, _) => Some(offset),
+ (None, _) => offset,
(Some(_), Some(state))
if state
.persisted_high_water
.is_some_and(|high_water| offset <= high_water) =>
{
- state.persisted_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?;
- Some(value)
+ value
+ }
+ (Some(path), None) => {
+ persist_offset_max(path, offset, self.consumer_offset_enforce_fsync).await?
}
- (Some(path), None) => Some(
- persist_offset_max(path, offset, self.consumer_offset_enforce_fsync)
- .await?,
- ),
};
self.durable_consumer_offsets.record_auto_commit(
pending.kind,
pending.consumer_id,
if tracked.is_none() {
- persisted_high_water.unwrap_or(offset)
+ persisted_high_water
} else {
offset
},
persisted_high_water,
);
- capacity.promote_to_durable(pending.consumer_id);
+ capacity.clear_stranded(pending.consumer_id);
+ if creates_group {
+ self.consumer_group_offsets_need_reconcile.set(true);
+ }
Ok(())
}
PendingConsumerOffsetMutation::Upsert(offset) => {
@@ -2175,7 +2185,10 @@ where
offset,
Some(offset),
);
- capacity.promote_to_durable(pending.consumer_id);
+ capacity.clear_stranded(pending.consumer_id);
+ if creates_group {
+ self.consumer_group_offsets_need_reconcile.set(true);
+ }
Ok(())
}
PendingConsumerOffsetMutation::Delete => {
@@ -2206,13 +2219,7 @@ where
offset: u64,
) -> bool {
self.durable_consumer_offsets
- .get(kind, consumer_id)
- .is_some_and(|state| {
- state.committed_offset >= offset
- && state
- .persisted_high_water
- .is_some_and(|high_water| offset <= high_water)
- })
+ .covers(kind, consumer_id, offset)
}
/// Reject completion from a removed partition or a view whose retained
@@ -2222,13 +2229,22 @@ where
applied.belongs_to(&self.durable_consumer_offsets)
&& self.observed_view == self.consensus.view()
&& !self.offset_reservations_need_resync.get()
- && !self
+ && (!self
.consumer_offset_capacity_for(applied.kind)
.is_uncertain()
+ || self
+ .durable_consumer_offsets
+ .contains(applied.kind, applied.consumer_id))
}
fn apply_consumer_offset_commit(&self, pending: PendingConsumerOffsetCommit) {
- if pending.kind == ConsumerKind::ConsumerGroup {
+ if pending.kind == ConsumerKind::ConsumerGroup
+ && (matches!(pending.mutation, PendingConsumerOffsetMutation::Delete)
+ || !self
+ .consumer_group_offsets
+ .pin()
+ .contains_key(&ConsumerGroupId(pending.consumer_id as usize)))
+ {
self.consumer_group_offsets_need_reconcile.set(true);
}
match pending.mutation {
@@ -2434,6 +2450,9 @@ where
&request_header,
committed_reply_body(request_header.operation),
);
+ if offset.is_some() {
+ self.release_consumer_offset_reservation(kind, consumer_id);
+ }
// Same rule as the committed path: a submit's waiter takes the reply,
// because `header.client` is then the VSR consensus id.
if let Some(waiter) = waiter {
@@ -2487,7 +2506,7 @@ where
&self,
kind: ConsumerKind,
consumer_id: u32,
- ) -> Result<CapacityReservation, ConsumerOffsetCapacityError> {
+ ) -> Result<(), ConsumerOffsetCapacityError> {
self.consumer_offset_capacity_for(kind)
.try_reserve(consumer_id, &self.durable_consumer_offsets)
}
@@ -2519,7 +2538,9 @@ where
consumer_id: u32,
waiter: &mut Option<consensus::Sender<Message<ReplyHeader>>>,
) -> bool {
- if self.consumer_offset_capacity_for(kind).is_uncertain() {
+ if self.consumer_offset_capacity_for(kind).is_uncertain()
+ && !self.durable_consumer_offsets.contains(kind, consumer_id)
+ {
Self::send_partition_deny_or_log(
&self.consensus,
header,
@@ -2578,7 +2599,6 @@ where
|| self
.consumer_offset_capacity_for(kind)
.is_stranded(consumer_id)
- || self.durable_consumer_offsets.contains(kind, consumer_id)
{
Ok(())
} else {
@@ -2593,12 +2613,22 @@ where
ReplicaLogContext::from_consensus(self.consensus(), PlaneKind::Partitions)
}
+ fn store_offset_range_error(&self, offset: u64) -> Option<IggyError> {
+ let current = self.stats.current_offset();
+ (offset > current || (current == 0 && self.stats.messages_count_inconsistent() == 0))
+ .then_some(IggyError::InvalidOffset(offset))
+ }
+
fn resynchronize_consumer_offset_reservations(&mut self) {
let current_view = self.consensus.view();
if current_view == self.observed_view && !self.offset_reservations_need_resync.get() {
return;
}
+ if current_view != self.observed_view {
+ self.queued_auto_commit_reservations.borrow_mut().clear();
+ }
+
let from_op = self
.consensus
.commit_min()
@@ -2665,7 +2695,29 @@ where
}
self.observed_view = current_view;
self.consumer_group_offsets_need_reconcile.set(true);
- self.offset_reservations_need_resync.set(decode_failed);
+ // Repair and truncation rearm this flag when journal contents change.
+ // A failed decode alone must not cause a full scan on every tick.
+ self.offset_reservations_need_resync.set(false);
+ if !decode_failed && self.consensus.is_primary() {
+ self.reclaim_phantom_offsets(ConsumerKind::Consumer);
+ self.reclaim_phantom_offsets(ConsumerKind::ConsumerGroup);
+ }
+ }
+
+ fn reclaim_phantom_offsets(&self, kind: ConsumerKind) {
+ let capacity = self.consumer_offset_capacity_for(kind);
+ let keep = |id| capacity.protects(id, &self.durable_consumer_offsets);
+ match kind {
+ ConsumerKind::Consumer => self
+ .consumer_offsets
+ .pin()
+ .retain(|id, _| u32::try_from(*id).ok().is_none_or(keep)),
+ ConsumerKind::ConsumerGroup => self
+ .consumer_group_offsets
+ .pin()
+ .retain(|id, _| u32::try_from(id.0).ok().is_none_or(keep)),
+ }
+ capacity.rearm_map_if_below_limit(self.consumer_offset_map_count(kind));
}
/// Build an owned [`PollPlan`] synchronously (no `.await`), so the caller
@@ -2724,6 +2776,30 @@ where
}
};
+ if args.auto_commit
+ && self.consensus.is_primary()
+ && let Ok(pending) = PendingConsumerOffsetCommit::try_from_polling_consumer(consumer, 0)
+ {
+ let capacity = self.consumer_offset_capacity_for(pending.kind);
+ if !capacity.is_uncertain()
+ && self.consumer_offset_map_count(pending.kind) >= capacity.limit()
+ {
+ let exists = match pending.kind {
+ ConsumerKind::Consumer => self
+ .consumer_offsets
+ .pin()
+ .contains_key(&(pending.consumer_id as usize)),
+ ConsumerKind::ConsumerGroup => self
+ .consumer_group_offsets
+ .pin()
+ .contains_key(&ConsumerGroupId(pending.consumer_id as usize)),
+ };
+ if !exists {
+ self.reclaim_phantom_offsets(pending.kind);
+ }
+ }
+ }
+
// Past the empty-return guards: only now build the auto-commit context,
// whose offset-path `format!()` is wasted on the early returns above.
let auto_commit = self.auto_commit_ctx(consumer, args.auto_commit);
@@ -3109,6 +3185,23 @@ 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| {
+ !self
+ .consumer_offset_capacity_for(reservation.kind)
+ .owns(reservation)
+ }) {
+ return;
+ }
// Taken by whichever arm answers: the deny paths, the NoAck fast path,
// or the pipeline entry that fires it at commit. Exactly one runs.
let mut reply = reply;
@@ -3303,30 +3396,26 @@ where
// code on this committed-shaped frame (op=commit_max) as success.
if matches!(message.header().operation, Operation::StoreConsumerOffset)
&& let Some((_, _, Some(requested_offset), _)) = consumer_offset
+ && let Some(error) = self.store_offset_range_error(requested_offset)
{
- let current_offset = self.stats.current_offset();
- let partition_empty =
- self.stats.messages_count_inconsistent() == 0 && current_offset == 0;
- if partition_empty || requested_offset > current_offset {
- emit_partition_diag(
- tracing::Level::WARN,
- &PartitionDiagEvent::new(
- ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions),
- "rejecting store_consumer_offset for out-of-range offset",
- )
- .with_operation(message.header().operation)
- .with_error(IggyError::InvalidOffset(requested_offset).to_string()),
- );
- Self::send_partition_deny_or_log(
- consensus,
- message.header(),
- IggyError::InvalidOffset(requested_offset).as_code(),
- "store_consumer_offset deny reply send failed",
- reply.take(),
+ emit_partition_diag(
+ tracing::Level::WARN,
+ &PartitionDiagEvent::new(
+ ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions),
+ "rejecting store_consumer_offset for out-of-range offset",
)
- .await;
- return;
- }
+ .with_operation(message.header().operation)
+ .with_error(error.to_string()),
+ );
+ Self::send_partition_deny_or_log(
+ consensus,
+ message.header(),
+ error.as_code(),
+ "store_consumer_offset deny reply send failed",
+ reply.take(),
+ )
+ .await;
+ return;
}
// The node-local fast path is safe only for a single-replica
@@ -3392,6 +3481,12 @@ where
waiter,
)
.await;
+ } else if let Some(reservation) = reservation.take() {
+ self.queued_auto_commit_reservations
+ .borrow_mut()
+ .entry((reservation.kind, reservation.consumer_id))
+ .or_default()
+ .push(reservation);
}
return;
}
@@ -3471,6 +3566,23 @@ where
while promoted < slots_freed {
let req = self.consensus().pop_queued_request();
let Some(mut req) = req else { break };
+ let _reservation =
+ Self::parse_consumer_offset_request(req.message.header().operation, &req.message)
+ .ok()
+ .and_then(|(kind, id, _, _)| {
+ if req.message.header().operation != Operation::StoreConsumerOffset
+ || !is_auto_commit_client(req.message.header().client)
+ {
+ return None;
+ }
+ let mut queued = self.queued_auto_commit_reservations.borrow_mut();
+ let reservations = queued.get_mut(&(kind, id))?;
+ let reservation = reservations.pop();
+ if reservations.is_empty() {
+ queued.remove(&(kind, id));
+ }
+ reservation
+ });
// Taken before the preflight so a refusal answers the parked waiter
// instead of waking it with `Canceled`.
@@ -3498,14 +3610,11 @@ where
.await;
continue;
};
- let current_offset = self.stats.current_offset();
- if offset > current_offset
- || (current_offset == 0 && self.stats.messages_count_inconsistent() == 0)
- {
+ if let Some(error) = self.store_offset_range_error(offset) {
Self::send_partition_deny_or_log(
self.consensus(),
req.message.header(),
- IggyError::InvalidOffset(offset).as_code(),
+ error.as_code(),
"queued offset range deny reply send failed",
reply_sender.take(),
)
@@ -5056,6 +5165,9 @@ where
send_fail_label: &'static str,
waiter: Option<consensus::Sender<Message<ReplyHeader>>>,
) {
+ if waiter.is_none() && is_auto_commit_client(header.client) {
+ return;
+ }
let reply = build_deny_reply_from_request(consensus, header, status);
Self::deliver_reply_or_log(consensus, header, reply, waiter, send_fail_label).await;
}
@@ -6105,10 +6217,7 @@ where
crate::state_transfer::strayed_offset_files(self.consumer_offsets_path.as_deref(), &[])
.into_iter()
.filter_map(|path| {
- std::path::Path::new(&path)
- .file_name()
- .and_then(|name| name.to_str())
- .and_then(|name| name.parse().ok())
+ crate::state_transfer::numeric_offset_id(&path)
.map(|id| (ConsumerKind::Consumer, id, path))
});
let strayed_groups = crate::state_transfer::strayed_offset_files(
@@ -6117,10 +6226,7 @@ where
)
.into_iter()
.filter_map(|path| {
- std::path::Path::new(&path)
- .file_name()
- .and_then(|name| name.to_str())
- .and_then(|name| name.parse().ok())
+ crate::state_transfer::numeric_offset_id(&path)
.map(|id| (ConsumerKind::ConsumerGroup, id, path))
});
for (kind, consumer_id, path) in consumer_paths
@@ -6175,6 +6281,7 @@ where
}
self.durable_consumer_offsets.clear();
self.pending_consumer_offset_commits.clear();
+ self.queued_auto_commit_reservations.borrow_mut().clear();
self.consumer_offset_capacity
.rebuild(&self.durable_consumer_offsets, std::iter::empty());
self.consumer_group_offset_capacity
@@ -8396,6 +8503,220 @@ mod tests {
);
}
+ #[compio::test]
+ async fn given_queued_auto_commit_when_view_changes_should_release_its_provisional_slot() {
+ let namespace = IggyNamespace::new(1, 1, 0);
+ let consensus = VsrConsensus::new(
+ TEST_CLUSTER,
+ 0,
+ 3,
+ namespace.inner(),
+ RecordingBus::default(),
+ LocalPipeline::with_capacities(1, 2),
+ );
+ consensus.init();
+ let mut partition: IggyPartition<RecordingBus> = IggyPartition::with_in_memory_storage(
+ Arc::new(PartitionStats::default()),
+ consensus,
+ IggyByteSize::from(1024 * 1024),
+ false,
+ );
+ partition.stats.increment_messages_count(1);
+ partition.set_consumer_offsets_max(2);
+ partition
+ .on_request(
+ store_offset_request(42, 1, ConsumerKind::Consumer, 7, 0, AckLevel::Quorum),
+ None,
+ )
+ .await;
+ let reservation = partition
+ .consumer_offset_capacity
+ .reserve_provisional(8, &partition.durable_consumer_offsets)
+ .unwrap();
+ partition
+ .on_request_with_reservation(
+ store_offset_request(
+ message_bus::AUTO_COMMIT_CLIENT_ID,
+ 1,
+ ConsumerKind::Consumer,
+ 8,
+ 0,
+ AckLevel::Quorum,
+ ),
+ None,
+ Some(reservation),
+ )
+ .await;
+ assert_eq!(
+ partition.occupied_consumer_offset_count(ConsumerKind::Consumer),
+ 2
+ );
+ assert_eq!(partition.queued_auto_commit_reservations.borrow().len(), 1);
+ partition.consensus.set_view(3);
+ partition.resynchronize_consumer_offset_reservations();
+ assert!(
+ partition
+ .queued_auto_commit_reservations
+ .borrow()
+ .is_empty()
+ );
+ assert_eq!(
+ partition.occupied_consumer_offset_count(ConsumerKind::Consumer),
+ 1
+ );
+ }
+
+ #[compio::test]
+ async fn given_auto_commit_guard_when_pump_admits_should_transfer_to_journal_without_reply() {
+ let (mut partition, sent) = recording_partition_at(0, 3);
+ partition.stats.increment_messages_count(1);
+ partition.set_consumer_offsets_max(1);
+ let reservation = partition
+ .consumer_offset_capacity
+ .reserve_provisional(7, &partition.durable_consumer_offsets)
+ .unwrap();
+ partition
+ .on_request_with_reservation(
+ store_offset_request(
+ message_bus::AUTO_COMMIT_CLIENT_ID,
+ 1,
+ ConsumerKind::Consumer,
+ 7,
+ 0,
+ AckLevel::Quorum,
+ ),
+ None,
+ Some(reservation),
+ )
+ .await;
+ assert_eq!(partition.pending_consumer_offset_commits.len(), 1);
+ assert_eq!(
+ partition.occupied_consumer_offset_count(ConsumerKind::Consumer),
+ 1
+ );
+ assert!(sent.borrow().is_empty());
+ partition
+ .apply_staged_consumer_offset_commit(1)
+ .await
+ .unwrap();
+ assert_eq!(
+ partition.durable_consumer_offset_count(ConsumerKind::Consumer),
+ 1
+ );
+ assert!(sent.borrow().is_empty());
+ }
+
+ #[compio::test]
+ async fn given_group_offset_updates_when_key_already_exists_should_keep_reconciliation_idle() {
+ let (mut partition, _) = recording_partition();
+ partition.stage_consumer_offset_upsert(1, ConsumerKind::ConsumerGroup, 7, 1, true);
+ partition
+ .apply_staged_consumer_offset_commit(1)
+ .await
+ .unwrap();
+ assert!(partition.consumer_group_offsets_reconcile_needed());
+ assert!(
+ partition
+ .dead_consumer_group_offset_ids(|_| true)
+ .is_empty()
+ );
+ assert!(!partition.consumer_group_offsets_reconcile_needed());
+ partition.stage_consumer_offset_upsert(2, ConsumerKind::ConsumerGroup, 7, 2, true);
+ partition
+ .apply_staged_consumer_offset_commit(2)
+ .await
+ .unwrap();
+ assert!(!partition.consumer_group_offsets_reconcile_needed());
+ partition.consumer_group_offsets.pin().insert(
+ ConsumerGroupId(8),
+ ConsumerOffset::new(ConsumerKind::ConsumerGroup, 8, 3, String::new()),
+ );
+ partition.stage_consumer_offset_upsert(3, ConsumerKind::ConsumerGroup, 8, 3, true);
+ partition
+ .apply_staged_consumer_offset_commit(3)
+ .await
+ .unwrap();
+ assert!(
+ partition.consumer_group_offsets_reconcile_needed(),
+ "first durable commit must discover an eager map key"
+ );
+ }
+
+ #[compio::test]
+ async fn given_primary_view_change_when_phantoms_exist_should_keep_only_protected_keys() {
+ let (mut partition, _) = recording_partition_at(0, 3);
+ partition.set_consumer_offsets_max(4);
+ partition.stats.increment_messages_count(1);
+ partition.seed_recovered_consumer_offset(ConsumerKind::Consumer, 7, 0, 0);
+ partition
+ .on_request(
+ store_offset_request(42, 1, ConsumerKind::Consumer, 8, 0, AckLevel::Quorum),
+ None,
+ )
+ .await;
+ let held = partition
+ .consumer_offset_capacity
+ .reserve_provisional(9, &partition.durable_consumer_offsets)
+ .unwrap();
+ for id in 7..=10 {
+ partition.consumer_offsets.pin().insert(
+ id as usize,
+ ConsumerOffset::new(ConsumerKind::Consumer, id, 0, String::new()),
+ );
+ }
+ partition.consensus.set_view(3);
+ partition.resynchronize_consumer_offset_reservations();
+ assert_eq!(partition.consumer_offsets.len(), 3);
+ assert!(!partition.consumer_offsets.pin().contains_key(&10));
+ for id in 7..=9 {
+ assert!(partition.consumer_offsets.pin().contains_key(&id));
+ }
+ drop(held);
+ }
+
+ #[test]
+ fn given_full_live_map_when_polling_existing_key_should_reclaim_only_for_new_keys() {
+ let (mut partition, _) = recording_partition();
+ partition.set_consumer_offsets_max(2);
+ partition.offset_space.committed_seeded = true;
+ partition.seed_recovered_consumer_offset(ConsumerKind::Consumer, 7, 0, 0);
+ for id in 7..=8 {
+ partition.consumer_offsets.pin().insert(
+ id as usize,
+ ConsumerOffset::new(ConsumerKind::Consumer, id, 0, String::new()),
+ );
+ }
+ let args = PollingArgs::new(iggy_common::PollingStrategy::first(), 1, true);
+ let _ = partition.build_poll_plan(PollingConsumer::Consumer(7, 0), &args, false);
+ assert_eq!(partition.consumer_offsets.len(), 2);
+ let _ = partition.build_poll_plan(PollingConsumer::Consumer(9, 0), &args, false);
+ assert_eq!(partition.consumer_offsets.len(), 1);
+ assert!(partition.consumer_offsets.pin().contains_key(&7));
+ }
+
+ #[test]
+ fn given_missing_retained_header_when_resync_fails_should_wait_for_new_journal_state() {
+ let (mut partition, _) = recording_partition();
+ partition.consensus.sequencer().set_sequence(1);
+ partition.consensus.set_view(1);
+ partition.resynchronize_consumer_offset_reservations();
+ assert!(partition.consumer_offset_capacity.is_uncertain());
+ assert!(!partition.offset_reservations_need_resync.get());
+ partition.seed_recovered_consumer_offset(ConsumerKind::Consumer, 7, 0, 0);
+ let existing = partition
+ .auto_commit_ctx(PollingConsumer::Consumer(7, 0), true)
+ .unwrap()
+ .apply(0)
+ .unwrap();
+ assert!(partition.auto_commit_admission_ready(&existing));
+ let new = partition
+ .auto_commit_ctx(PollingConsumer::Consumer(8, 0), true)
+ .unwrap()
+ .apply(0)
+ .unwrap();
+ assert!(!partition.auto_commit_admission_ready(&new));
+ }
+
#[compio::test]
async fn given_full_offset_table_when_storing_new_id_should_reject_before_replication() {
let (mut partition, sent_to_clients) = recording_partition();
diff --git a/core/partitions/src/iggy_partitions.rs b/core/partitions/src/iggy_partitions.rs
index 2a2ee6e1e..d66b96dc7 100644
--- a/core/partitions/src/iggy_partitions.rs
+++ b/core/partitions/src/iggy_partitions.rs
@@ -628,6 +628,22 @@ where
let _ = waiter.send(build_deny_reply_from_request_header(header, status));
}
}
+
+ pub async fn on_auto_commit_request(
+ &self,
+ request: Message<RoutedRequestHeader>,
+ reservation: crate::AutoCommitReservation,
+ ) {
+ let namespace = IggyNamespace::from_raw(request.header().group);
+ if self.is_tombstoned(&namespace) {
+ return;
+ }
+ if let Some(partition) = self.get_mut_by_ns(&namespace) {
+ partition
+ .on_request_with_reservation(request, None, Some(reservation))
+ .await;
+ }
+ }
}
impl<B, SB> Plane<VsrConsensus<B>> for IggyPartitions<B, SB>
diff --git a/core/partitions/src/poll_plan.rs b/core/partitions/src/poll_plan.rs
index 176fc8245..51ae28a28 100644
--- a/core/partitions/src/poll_plan.rs
+++ b/core/partitions/src/poll_plan.rs
@@ -311,6 +311,9 @@ impl PollPlan {
/// # Errors
/// Returns a capacity error when auto-commit would create a new live-map
/// entry after the configured per-kind bound has been reached.
+ /// The serving shard also rejects completion with `TransientNotAccepted`
+ /// if the partition disappeared or changed incarnation during disk I/O.
+ /// No fragments are returned for that rejected completion.
pub async fn execute(
self,
) -> Result<(PollFragments<4096>, u64, Option<AutoCommitApplied>), ConsumerOffsetCapacityError>
@@ -890,64 +893,37 @@ impl AutoCommitCtx {
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;
- let guard = map.pin();
- if let Some(existing) = guard.get(&(consumer_id as usize)) {
- Some(existing.offset.fetch_max(offset, Ordering::Relaxed))
- } else {
- self.capacity.admit_local_map_key(guard.len())?;
- let created = create_path.as_deref().map_or_else(
- || {
- ConsumerOffset::new(
- ConsumerKind::Consumer,
- consumer_id,
- offset,
- String::new(),
- )
- },
- |path| ConsumerOffset::default_for_consumer(consumer_id, path),
- );
- created.offset.store(offset, Ordering::Relaxed);
- guard.insert(consumer_id as usize, created);
- None
- }
- }
+ } => apply_local_offset(
+ offsets,
+ *consumer_id as usize,
+ offset,
+ &self.capacity,
+ || 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;
- let guard = map.pin();
- if let Some(existing) = guard.get(&key) {
- Some(existing.offset.fetch_max(offset, Ordering::Relaxed))
- } else {
- self.capacity.admit_local_map_key(guard.len())?;
- let created = create_path.as_deref().map_or_else(
- || {
- ConsumerOffset::new(
- ConsumerKind::ConsumerGroup,
- group_id,
- offset,
- String::new(),
- )
- },
- |path| ConsumerOffset::default_for_consumer_group(key, path),
- );
- created.offset.store(offset, Ordering::Relaxed);
- guard.insert(key, created);
- None
- }
- }
+ } => apply_local_offset(
+ offsets,
+ ConsumerGroupId(*group_id as usize),
+ offset,
+ &self.capacity,
+ || create(create_path.as_deref()),
+ )?,
};
Ok(AutoCommitApplied {
kind,
@@ -980,13 +956,7 @@ impl AutoCommitApplied {
) -> Result<Option<AutoCommitReservation>, ConsumerOffsetCapacityError> {
if self
.durable
- .get(self.kind, self.consumer_id)
- .is_some_and(|state| {
- state.committed_offset >= self.offset
- && state
- .persisted_high_water
- .is_some_and(|high_water| self.offset <= high_water)
- })
+ .covers(self.kind, self.consumer_id, self.offset)
{
return Ok(None);
}
@@ -1007,35 +977,51 @@ impl AutoCommitApplied {
offsets,
consumer_id,
..
- } => {
- let guard = offsets.pin();
- if let Some(previous) = self.previous_offset {
- if let Some(entry) = guard.get(&(*consumer_id as usize)) {
- entry.offset.store(previous, Ordering::Relaxed);
- }
- } else {
- guard.remove(&(*consumer_id as usize));
- }
- guard.len()
- }
+ } => rollback_local_offset(offsets, *consumer_id as usize, self.previous_offset),
AutoCommitTarget::ConsumerGroup {
offsets, group_id, ..
- } => {
- let guard = offsets.pin();
- if let Some(previous) = self.previous_offset {
- if let Some(entry) = guard.get(&ConsumerGroupId(*group_id as usize)) {
- entry.offset.store(previous, Ordering::Relaxed);
- }
- } else {
- guard.remove(&ConsumerGroupId(*group_id as usize));
- }
- guard.len()
- }
+ } => rollback_local_offset(
+ offsets,
+ ConsumerGroupId(*group_id as usize),
+ self.previous_offset,
+ ),
};
self.capacity.rearm_map_if_below_limit(map_len);
}
}
+fn rollback_local_offset<K: Hash + Eq + Send + Sync + Copy>(
+ map: &papaya::HashMap<K, ConsumerOffset>,
+ key: K,
+ previous: Option<u64>,
+) -> usize {
+ 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);
+ }
+ guard.len()
+}
+
+fn apply_local_offset<K: Hash + Eq + Clone + Send + Sync>(
+ map: &papaya::HashMap<K, ConsumerOffset>,
+ key: K,
+ offset: u64,
+ capacity: &ConsumerOffsetCapacity,
+ 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())?;
+ guard.insert(key, create());
+ Ok(None)
+}
+
/// Upsert a committed offset into a lock-free `papaya` offset map: bump an
/// existing entry in place, or build one via `create_on_miss` on first commit
/// for a consumer/group that has none yet. Shared by the pump's
diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs
index ff5efe345..41d7f6f09 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -2000,12 +2000,9 @@ where
let mut consumers = self
.durable_consumer_offsets
.committed_entries(ConsumerKind::Consumer);
+ let consumer_map = self.consumer_offsets.pin();
for (consumer_id, _) in &consumers {
- if !self
- .consumer_offsets
- .pin()
- .contains_key(&(*consumer_id as usize))
- {
+ if !consumer_map.contains_key(&(*consumer_id as usize)) {
return Err(
PartitionTransferUnavailable::ConsumerOffsetStateInconsistent {
kind: ConsumerKind::Consumer,
@@ -2018,12 +2015,9 @@ where
let mut groups = self
.durable_consumer_offsets
.committed_entries(ConsumerKind::ConsumerGroup);
+ let group_map = self.consumer_group_offsets.pin();
for (consumer_id, _) in &groups {
- if !self
- .consumer_group_offsets
- .pin()
- .contains_key(&ConsumerGroupId(*consumer_id as usize))
- {
+ if !group_map.contains_key(&ConsumerGroupId(*consumer_id as usize)) {
return Err(
PartitionTransferUnavailable::ConsumerOffsetStateInconsistent {
kind: ConsumerKind::ConsumerGroup,
@@ -2825,6 +2819,7 @@ where
}
self.durable_consumer_offsets.clear();
self.pending_consumer_offset_commits.clear();
+ self.queued_auto_commit_reservations.borrow_mut().clear();
self.consumer_offset_capacity
.rebuild(&self.durable_consumer_offsets, std::iter::empty());
self.consumer_group_offset_capacity
@@ -3450,7 +3445,7 @@ pub(crate) fn strayed_offset_files(dir: Option<&str>, incoming: &[(u32, u64)]) -
.collect()
}
-fn numeric_offset_id(path: &str) -> Option<u32> {
+pub(crate) fn numeric_offset_id(path: &str) -> Option<u32> {
Path::new(path).file_name()?.to_str()?.parse().ok()
}
diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml
index 09ef84d03..93993589c 100644
--- a/core/sdk/Cargo.toml
+++ b/core/sdk/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy"
-version = "0.11.0-edge.6"
+version = "0.11.0-edge.7"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
rust-version.workspace = true
diff --git a/core/sdk/src/clients/consumer.rs b/core/sdk/src/clients/consumer.rs
index 1f38f2c76..e957cce7a 100644
--- a/core/sdk/src/clients/consumer.rs
+++ b/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
+/// its consumer needs a new offset key at the partition's configured limit.
+/// The rejected poll returns no messages. Existing offset keys remain usable,
+/// and polling with server-side auto-commit disabled allocates no offset key.
pub struct IggyConsumer {
initialized: bool,
shutdown: Arc<AtomicBool>,
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index 1d5911488..553d8d9ff 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "server"
-version = "0.9.0-edge.6"
+version = "0.9.0-edge.7"
edition = "2024"
license = "Apache-2.0"
publish = false
diff --git a/core/server/config.toml b/core/server/config.toml
index 285e9c807..909b791c5 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -1027,6 +1027,9 @@ 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.
# Must be > 0 and <= 262144.
consumer_offsets_max = 4096
diff --git a/core/server/src/boot/recovery.rs b/core/server/src/boot/recovery.rs
index 0610e826a..917306bdc 100644
--- a/core/server/src/boot/recovery.rs
+++ b/core/server/src/boot/recovery.rs
@@ -336,7 +336,7 @@ const _: () = assert!(
== partitions::DEFAULT_CONSUMER_OFFSETS_MAX
);
const _: () = assert!(
- 2 * configs::partition::PARTITION_CONSUMER_OFFSETS_CEILING
+ 4 * configs::partition::PARTITION_CONSUMER_OFFSETS_CEILING
<= partitions::CONSUMER_OFFSETS_ENTRIES_MAX as usize
);
const _: () =
diff --git a/core/server/src/consumer_group.rs b/core/server/src/consumer_group.rs
index 2f347782f..1a6beed9b 100644
--- a/core/server/src/consumer_group.rs
+++ b/core/server/src/consumer_group.rs
@@ -25,7 +25,7 @@
//! primary enriches the op here before replication, mirroring the PAT mint
//! in [`crate::pat`] and the password hash in [`crate::users`].
-use crate::responses::resolve_partition_namespace;
+use crate::responses::{missing_consumer_group_error, resolve_partition_namespace};
use crate::shell::{ShellBus, ShellShard};
use crate::wire::{request_body, rewrite_request_body};
use consensus::MetadataHandle;
@@ -235,9 +235,12 @@ where
if wire.consumer.kind != KIND_CONSUMER_GROUP {
return Ok(request);
}
- let group_id =
- resolve_group_offset_id(shard, &wire.consumer, (&wire.stream_id, &wire.topic_id))
- .ok_or(IggyError::InvalidIdentifier)?;
+ let group_id = resolve_group_offset_id(
+ shard,
+ &wire.consumer,
+ (&wire.stream_id, &wire.topic_id),
+ )
+ .ok_or_else(|| missing_consumer_group_error(&wire.consumer.id, &wire.topic_id))?;
// The partition-plane group-offset key is u32 (see the documented
// ceiling on `Topic::next_consumer_group_id`). Clamp on the
// ~4-billion-creates overflow rather than panic this live
@@ -270,9 +273,6 @@ where
S: 'static,
SB: SuperblockStore + 'static,
{
- if consumer.kind != KIND_CONSUMER_GROUP {
- return None;
- }
shard
.plane
.metadata()
diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs
index 1c8e21711..48bbe0b5b 100644
--- a/core/server/src/dispatch/partition.rs
+++ b/core/server/src/dispatch/partition.rs
@@ -51,7 +51,7 @@ use iggy_binary_protocol::PrepareHeader;
use iggy_binary_protocol::primitives::consumer::WireConsumer;
use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy;
use iggy_binary_protocol::requests::consumer_offsets::{
- DeleteConsumerOffsetRequest, GetConsumerOffsetRequest, StoreConsumerOffsetRequest,
+ GetConsumerOffsetRequest, StoreConsumerOffsetRequest,
};
use iggy_binary_protocol::requests::messages::PollMessagesRequest;
use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest;
@@ -128,7 +128,7 @@ where
Err(error) => {
shard.metrics().record_consumer_offset_denied(error.kind);
warn_auto_commit_capacity(namespace, error);
- PartitionReadReply::Rejected(IggyError::TooManyConsumerOffsets)
+ PartitionReadReply::Rejected(error.into())
}
};
let _ = reply.try_send(result);
@@ -228,7 +228,7 @@ fn spawn_poll_io<B, MJ, S, SB>(
Err(error) => {
shard.metrics().record_consumer_offset_denied(error.kind);
warn_auto_commit_capacity(namespace, error);
- PartitionReadReply::Rejected(IggyError::TooManyConsumerOffsets)
+ PartitionReadReply::Rejected(error.into())
}
};
let _ = reply.try_send(result);
@@ -296,7 +296,7 @@ where
shard.metrics().record_consumer_offset_denied(applied.kind);
warn_auto_commit_capacity(namespace, error);
applied.rollback_created();
- return Err(IggyError::TooManyConsumerOffsets);
+ return Err(error.into());
}
};
let message = match build_auto_commit_request(namespace, applied) {
@@ -313,15 +313,13 @@ where
};
// Routes by namespace to this same owning primary shard's inbox. The pump
// admits it next turn exactly like a client store. `dispatch` never blocks.
- let Ok(ticket) = shard.partition_submit(namespace, message) else {
+ if shard
+ .submit_auto_commit_offset(message, reservation)
+ .is_err()
+ {
applied.rollback_created();
return Err(IggyError::TransientNotAccepted);
- };
- let shard = Rc::clone(shard);
- shard.bus.clone().spawn(async move {
- let _ = shard.await_partition_submit(ticket).await;
- drop(reservation);
- });
+ }
applied.mark_served();
Ok(())
}
@@ -449,7 +447,10 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>(
operation = ?header.operation,
"partition request with unresolved namespace; replying denied"
);
- let status = if error == IggyError::InvalidIdentifier {
+ let status = if matches!(
+ error,
+ IggyError::ConsumerGroupIdNotFound(..) | IggyError::ConsumerGroupNameNotFound(..)
+ ) {
error.as_code()
} else {
IggyError::ResourceNotFound(String::new()).as_code()
@@ -650,22 +651,12 @@ async fn relay_partition_reply<B, MJ, S, SB>(
}
fn consumer_offset_kind(request: &Message<RoutedRequestHeader>) -> Option<ConsumerKind> {
- let kind = match request.header().operation {
- Operation::StoreConsumerOffset => {
- StoreConsumerOffsetRequest::decode_from(request_body(request))
- .ok()?
- .consumer
- .kind
- }
- Operation::DeleteConsumerOffset => {
- DeleteConsumerOffsetRequest::decode_from(request_body(request))
- .ok()?
- .consumer
- .kind
- }
- _ => return None,
- };
- ConsumerKind::from_code(kind).ok()
+ if request.header().operation != Operation::StoreConsumerOffset {
+ return None;
+ }
+ // WireConsumer starts with its kind byte. The dispatch path has already
+ // decoded and validated the complete request.
+ ConsumerKind::from_code(*request_body(request).first()?).ok()
}
/// Serve `poll_messages`: resolve the partition namespace, run the read on
diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs
index b8fc36c3d..6c3f888c7 100644
--- a/core/server/src/partition_helpers.rs
+++ b/core/server/src/partition_helpers.rs
@@ -276,10 +276,16 @@ pub fn configure_consumer_offsets(
enforce_fsync,
);
for consumer_id in numeric_offset_file_ids(&consumer_offsets_path) {
- partition.seed_stranded_consumer_offset(ConsumerKind::Consumer, consumer_id);
+ if partition.seed_stranded_consumer_offset(ConsumerKind::Consumer, consumer_id) {
+ warn!(stream_id, topic_id, partition_id, consumer_id, path = %consumer_offsets_path,
+ "unloaded consumer offset file retains its capacity slot until updated or deleted");
+ }
}
for group_id in numeric_offset_file_ids(&consumer_group_offsets_path) {
- partition.seed_stranded_consumer_offset(ConsumerKind::ConsumerGroup, group_id);
+ if partition.seed_stranded_consumer_offset(ConsumerKind::ConsumerGroup, group_id) {
+ warn!(stream_id, topic_id, partition_id, group_id, path = %consumer_group_offsets_path,
+ "unloaded group offset file retains its capacity slot until repaired or reclaimed");
+ }
}
for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] {
let count = partition.occupied_consumer_offset_count(kind);
diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs
index 21f018085..8fedc7204 100644
--- a/core/server/src/partition_reconciler.rs
+++ b/core/server/src/partition_reconciler.rs
@@ -194,6 +194,7 @@ use tracing::{debug, error, trace};
const BACKOFF_BASE: Duration = Duration::from_secs(1);
const BACKOFF_MAX: Duration = Duration::from_mins(1);
+const GROUP_OFFSET_DELETES_PER_PASS: usize = 32;
/// Consecutive same-cause failures before [`ReconcilerCtx::record_failure`]
/// escalates to an operator-visible error (the backoff is capped, so
@@ -235,6 +236,8 @@ pub struct ReconcilerCtx {
/// `true` when the previous pass made no changes. Only then is a
/// same-`revision` pass safe to skip.
last_pass_noop: Cell<bool>,
+ group_offset_cleanup_inflight: Rc<RefCell<AHashSet<IggyNamespace>>>,
+ group_offset_cleanup_completed: Rc<Cell<usize>>,
}
impl ReconcilerCtx {
@@ -257,6 +260,8 @@ impl ReconcilerCtx {
failure_state: RefCell::new(AHashMap::new()),
last_revision: Cell::new(None),
last_pass_noop: Cell::new(false),
+ group_offset_cleanup_inflight: Rc::new(RefCell::new(AHashSet::new())),
+ group_offset_cleanup_completed: Rc::new(Cell::new(0)),
}
}
@@ -389,11 +394,11 @@ struct PassCounters {
backoff_skipped: usize,
/// Stale incarnations (slab-key reuse) torn down for rebuild.
stale: usize,
- /// Consumer-group offsets reclaimed for groups deleted while their topic
- /// survived (a bare `DeleteConsumerGroup`, not a topic/stream delete).
- cg_offsets_purged: usize,
- /// Consumer-group offset files whose unlink failed and remain queued for
- /// the next pass. Counted so the revision fast-skip cannot strand them.
+ /// Group-offset deletes successfully handed to the pump this pass.
+ cg_offsets_submitted: usize,
+ /// Successful replicated deletes reported by detached completion tasks.
+ cg_offsets_completed: usize,
+ /// Group-offset deletes refused by the inbox and needing another pass.
cg_offsets_pending: usize,
/// Committed delete watermarks not yet fully enforced on local segments.
/// Counted so the pass does not arm the fast-skip: the pump can be
@@ -432,7 +437,8 @@ impl PassCounters {
+ self.removed_routed
+ self.backoff_skipped
+ self.stale
- + self.cg_offsets_purged
+ + self.cg_offsets_submitted
+ + self.cg_offsets_completed
+ self.cg_offsets_pending
+ self.trims_pending
+ self.purges_staged
@@ -474,6 +480,7 @@ 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| {
ctx.shard
@@ -481,6 +488,10 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool {
.partitions()
.with_partition(namespace, |partition| {
partition.consumer_group_offsets_reconcile_needed()
+ && !ctx
+ .group_offset_cleanup_inflight
+ .borrow()
+ .contains(namespace)
})
.unwrap_or(false)
})
@@ -494,12 +505,15 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool {
let target = snapshot_target_namespaces(ctx);
let target_set: AHashSet<IggyNamespace> = target.iter().map(|partition| partition.ns).collect();
- let mut counters = PassCounters::default();
+ let mut counters = PassCounters {
+ cg_offsets_completed: ctx.group_offset_cleanup_completed.replace(0),
+ ..PassCounters::default()
+ };
reconcile_additions(ctx, target, &mut counters).await;
reconcile_removals(ctx, &target_set, &mut counters).await;
reconcile_parked_frames(ctx, &mut counters);
- reconcile_consumer_group_offsets(ctx, &mut counters).await;
+ reconcile_consumer_group_offsets(ctx, &mut counters);
reconcile_segment_truncations(ctx, &mut counters);
reconcile_partition_purges(ctx, &mut counters);
@@ -982,11 +996,18 @@ async fn tear_down_owned_partition(
/// Reclaim deleted groups through ordered offset deletes. Replicas must see
/// each delete before a replacement store can reuse its durable slot.
-async fn reconcile_consumer_group_offsets(ctx: &ReconcilerCtx, counters: &mut PassCounters) {
+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 namespace in owned {
+ if ctx
+ .group_offset_cleanup_inflight
+ .borrow()
+ .contains(&namespace)
+ {
+ continue;
+ }
let live = live_groups.get(&(namespace.stream_id(), namespace.topic_id()));
let dead = partitions
.with_partition(&namespace, |partition| {
@@ -997,32 +1018,48 @@ async fn reconcile_consumer_group_offsets(ctx: &ReconcilerCtx, counters: &mut Pa
.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(32));
- for consumer_id in dead.into_iter().take(32) {
- counters.cg_offsets_pending += 1;
+ 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) {
let request = group_offset_delete_request(namespace, consumer_id);
if let Ok(ticket) = ctx.shard.partition_submit(namespace, request) {
+ counters.cg_offsets_submitted += 1;
tickets.push(ticket);
+ } else {
+ counters.cg_offsets_pending += 1;
}
}
- let replies = futures::future::join_all(
- tickets
- .into_iter()
- .map(|ticket| ctx.shard.await_partition_submit(ticket)),
- )
- .await;
- for reply in replies {
- let Some(reply) = reply else {
- continue;
- };
- let header = reply
- .as_slice()
- .get(..size_of::<ReplyHeader>())
- .and_then(|bytes| bytemuck::checked::try_from_bytes::<ReplyHeader>(bytes).ok());
- if header.is_some_and(|header| header.status == 0) {
- counters.cg_offsets_purged += 1;
- }
+ if tickets.is_empty() {
+ continue;
}
+ ctx.group_offset_cleanup_inflight
+ .borrow_mut()
+ .insert(namespace);
+ let inflight = Rc::clone(&ctx.group_offset_cleanup_inflight);
+ let completed = Rc::clone(&ctx.group_offset_cleanup_completed);
+ let shard = Rc::clone(&ctx.shard);
+ shard.bus.clone().spawn(async move {
+ let replies = futures::future::join_all(
+ tickets
+ .into_iter()
+ .map(|ticket| shard.await_partition_submit(ticket)),
+ )
+ .await;
+ let count = replies
+ .into_iter()
+ .flatten()
+ .filter(|reply| {
+ reply
+ .as_slice()
+ .get(..size_of::<ReplyHeader>())
+ .and_then(|bytes| {
+ bytemuck::checked::try_from_bytes::<ReplyHeader>(bytes).ok()
+ })
+ .is_some_and(|header| header.status == 0)
+ })
+ .count();
+ completed.set(completed.get() + count);
+ inflight.borrow_mut().remove(&namespace);
+ });
}
}
@@ -1331,8 +1368,9 @@ pub fn install_tick_handler(shard: &Rc<ServerShard>, wake_tx: WakeTx) {
#[cfg(test)]
mod tests {
use super::{
- FailureCause, FailureRecord, ReconcilerCtx, build_partition_fresh,
- delete_partitions_from_disk, fetch_partition_stats, reconcile_once,
+ FailureCause, FailureRecord, PassCounters, ReconcilerCtx, build_partition_fresh,
+ delete_partitions_from_disk, fetch_partition_stats, reconcile_consumer_group_offsets,
+ reconcile_once,
};
use configs::server::{ServerConfig, ServerSystemConfig};
use consensus::{MetadataHandle, PartitionsHandle};
@@ -3241,6 +3279,60 @@ mod tests {
/// Failed delivery must leave the offset and its quota slot intact for a
/// later replicated delete. This fixture deliberately has no running pump.
+ #[compio::test]
+ async fn given_pending_group_cleanup_when_another_topic_arrives_should_reconcile_without_waiting()
+ {
+ let tmp = TempDir::new().unwrap();
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "cleanup-stream");
+ seed_topic(&mux, 2, 0, "cleanup-topic", vec![assignment(0, 1)]);
+ let (shard, _inbox) = build_test_shard_with_inbox(0, &config, mux, 32);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+ reconcile_pass(&ctx).await;
+ let namespace = IggyNamespace::new(0, 0, 0);
+ shard
+ .plane
+ .partitions()
+ .with_partition(&namespace, |partition| {
+ partition.consumer_group_offsets.pin().insert(
+ iggy_common::ConsumerGroupId(7),
+ iggy_common::ConsumerOffset::new(
+ iggy_common::ConsumerKind::ConsumerGroup,
+ 7,
+ 0,
+ String::new(),
+ ),
+ );
+ });
+ reconcile_consumer_group_offsets(&ctx, &mut PassCounters::default());
+ assert!(
+ ctx.group_offset_cleanup_inflight
+ .borrow()
+ .contains(&namespace)
+ );
+ seed_topic(
+ &shard.plane.metadata().mux_stm,
+ 3,
+ 0,
+ "unrelated-topic",
+ vec![assignment(0, 1)],
+ );
+ reconcile_pass(&ctx).await;
+ assert!(
+ shard
+ .plane
+ .partitions()
+ .contains(&IggyNamespace::new(0, 1, 0)),
+ "an unanswered cleanup must not block partition creation"
+ );
+ assert!(
+ ctx.group_offset_cleanup_inflight
+ .borrow()
+ .contains(&namespace)
+ );
+ }
+
#[compio::test]
async fn given_deleted_group_when_cleanup_submit_fails_should_preserve_state_for_retry() {
use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset};
diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs
index cc0654af1..2b9b524a5 100644
--- a/core/server/src/responses.rs
+++ b/core/server/src/responses.rs
@@ -279,12 +279,6 @@ where
}
let partition_id = partition_id.ok_or(IggyError::InvalidIdentifier)?;
let streams = shard.plane.metadata().mux_stm.streams();
- if streams
- .resolve_consumer_group_id(stream_id, topic_id, &consumer.id)
- .is_none()
- {
- return Err(IggyError::InvalidIdentifier);
- }
#[allow(clippy::cast_possible_truncation)]
streams
// Commit fence: allow a pending-revoked partition (the source commits it
@@ -298,10 +292,28 @@ where
false,
)
.map(|_| ())
- .ok_or(IggyError::ConsumerGroupPartitionNotOwned(
- client_id as u32,
- partition_id,
- ))
+ .ok_or_else(|| {
+ if streams
+ .resolve_consumer_group_id(stream_id, topic_id, &consumer.id)
+ .is_none()
+ {
+ missing_consumer_group_error(&consumer.id, topic_id)
+ } else {
+ IggyError::ConsumerGroupPartitionNotOwned(client_id as u32, partition_id)
+ }
+ })
+}
+
+pub fn missing_consumer_group_error(group: &WireIdentifier, topic: &WireIdentifier) -> IggyError {
+ let topic = wire_identifier_for_display(topic);
+ match group {
+ WireIdentifier::Numeric(_) => {
+ IggyError::ConsumerGroupIdNotFound(wire_identifier_for_display(group), topic)
+ }
+ WireIdentifier::String(name) => {
+ IggyError::ConsumerGroupNameNotFound(name.as_str().to_owned(), topic)
+ }
+ }
}
/// Fence a consumer-group offset op then resolve its target partition
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 4dbcdf428..2922b8150 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -114,6 +114,7 @@ where
pub struct PartitionMaterialisation {
epoch: u64,
created_view: u32,
+ consumer_offsets_max: usize,
}
#[cfg(feature = "simulator")]
@@ -123,8 +124,15 @@ impl PartitionMaterialisation {
Self {
epoch,
created_view,
+ consumer_offsets_max: partitions::DEFAULT_CONSUMER_OFFSETS_MAX,
}
}
+
+ #[must_use]
+ pub const fn with_consumer_offsets_max(mut self, consumer_offsets_max: usize) -> Self {
+ self.consumer_offsets_max = consumer_offsets_max;
+ self
+ }
}
/// Replica id + count bundle.
@@ -736,6 +744,12 @@ pub enum LifecycleFrame {
request: Message<RoutedRequestHeader>,
reply: Sender<Option<Message<GenericHeader>>>,
},
+ /// Local auto-commit submission. The guard travels with the frame so an
+ /// inbox drop or admission refusal releases its provisional key directly.
+ AutoCommitSubmit {
+ request: Message<RoutedRequestHeader>,
+ reservation: partitions::AutoCommitReservation,
+ },
/// Shard 0 broadcasts after a partition-shaped metadata commit; wakes
/// the per-shard reconciler. No payload: reconciler re-reads target
/// state. Drops covered by the periodic safety tick.
@@ -2010,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,
+ crate::coordinator::classify_try_send_err(&error),
+ );
+ PartitionSubmitRefused
+ })
+ }
+
/// Wait out a submitted write's committed reply.
///
/// `None` = reply channel dropped before a reply (view-change reset, park
@@ -3966,7 +4006,6 @@ where
// and `adopt_retained_log` are configured out and the crate does not compile.
// The feature forwards to `partitions/simulator` instead.
#[cfg(feature = "simulator")]
- #[allow(clippy::too_many_arguments)]
pub fn init_partition(
&self,
namespace: IggyNamespace,
@@ -3975,7 +4014,6 @@ where
retained: Option<partitions::RetainedPartitionState>,
restore_frontier: bool,
materialisation: PartitionMaterialisation,
- consumer_offsets_max: usize,
) where
B: MessageBus + Clone + 'static,
T: ShardsTable,
@@ -3983,6 +4021,7 @@ where
let PartitionMaterialisation {
epoch,
created_view,
+ consumer_offsets_max,
} = materialisation;
let partitions = self.plane.partitions();
if partitions.contains(&namespace) {
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
index d355b0381..91b6a2001 100644
--- a/core/shard/src/metrics.rs
+++ b/core/shard/src/metrics.rs
@@ -189,6 +189,13 @@ fn reason_index(s: &str) -> Option<usize> {
REASONS.iter().position(|r| *r == s)
}
+const fn consumer_kind_index(kind: ConsumerKind) -> usize {
+ match kind {
+ ConsumerKind::Consumer => 0,
+ ConsumerKind::ConsumerGroup => 1,
+ }
+}
+
/// Per-shard metric handles.
///
/// Cheap to clone (`Arc` of a `Family` under the hood). Each shard owns
@@ -271,21 +278,13 @@ impl ShardMetrics {
}
pub fn record_consumer_offset_denied(&self, kind: ConsumerKind) {
- let index = match kind {
- ConsumerKind::Consumer => 0,
- ConsumerKind::ConsumerGroup => 1,
- };
- self.consumer_offset_denied_counters[index].inc();
+ self.consumer_offset_denied_counters[consumer_kind_index(kind)].inc();
}
- #[cfg(any(test, feature = "simulator"))]
+ #[cfg(test)]
#[must_use]
pub fn consumer_offset_denied_value(&self, kind: ConsumerKind) -> u64 {
- let index = match kind {
- ConsumerKind::Consumer => 0,
- ConsumerKind::ConsumerGroup => 1,
- };
- self.consumer_offset_denied_counters[index].get()
+ self.consumer_offset_denied_counters[consumer_kind_index(kind)].get()
}
/// Bumped every time a client request is answered with a retryable denial
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index 4a3a0d214..b7465d2b7 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -728,6 +728,15 @@ where
// already made.
self.on_partition_submit(request, reply).await;
}
+ LifecycleFrame::AutoCommitSubmit {
+ request,
+ reservation,
+ } => {
+ self.plane
+ .partitions()
+ .on_auto_commit_request(request, reservation)
+ .await;
+ }
LifecycleFrame::MetadataCommitTick => {
// Reconciler may not yet be wired (e.g. mid-bootstrap, or
// single-shard tests that never enable the reconciler loop).
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index a4e055c82..242d4004f 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -505,11 +505,6 @@ impl Simulator {
self.consumer_offsets_max = consumer_offsets_max;
}
- #[must_use]
- pub const fn consumer_offsets_max(&self) -> usize {
- self.consumer_offsets_max
- }
-
#[must_use]
/// # Panics
/// Panics when `replica_idx` is outside the simulated roster.
@@ -1350,8 +1345,10 @@ impl Simulator {
};
// Partitions are driven directly, so a poll's auto-commit is never
// replicated (the serving shard's job in the real server). Offset discarded.
- let (fragments, _commit_offset, _auto_commit) = futures::executor::block_on(plan.execute())
- .map_err(|_| IggyError::TooManyConsumerOffsets)?;
+ let (fragments, _commit_offset, auto_commit) = futures::executor::block_on(plan.execute())?;
+ if let Some(applied) = auto_commit {
+ applied.mark_served();
+ }
Ok(fragments)
}
@@ -1527,8 +1524,8 @@ fn materialise_partition(
recovered_state,
retained,
restore_frontier,
- PartitionMaterialisation::new(epoch, created_view),
- consumer_offsets_max,
+ PartitionMaterialisation::new(epoch, created_view)
+ .with_consumer_offsets_max(consumer_offsets_max),
);
for shard in &replica.shards {
shard.shards_table().insert(
@@ -3059,7 +3056,6 @@ mod tests {
None,
false,
PartitionMaterialisation::new(0, 0),
- partitions::DEFAULT_CONSUMER_OFFSETS_MAX,
);
});
executor.run_until_stalled(POLL_BUDGET); // grow while the borrow is live
@@ -3102,7 +3098,6 @@ mod tests {
None,
false,
PartitionMaterialisation::new(0, 0),
- partitions::DEFAULT_CONSUMER_OFFSETS_MAX,
);
});
executor.run_until_stalled(POLL_BUDGET);
diff --git a/core/simulator/src/workload/effect.rs b/core/simulator/src/workload/effect.rs
index a9d8c868c..e8df466e9 100644
--- a/core/simulator/src/workload/effect.rs
+++ b/core/simulator/src/workload/effect.rs
@@ -17,7 +17,7 @@
//! Predicted server-state mutations emitted by op modules on commit.
//!
-//! Name-keyed throughout. Server-ng emits empty reply bodies, so the
+//! Name-keyed throughout. Server emits empty reply bodies, so the
//! workload cannot recover server-assigned numeric ids; shadow lookups
//! address entities by name (`WireIdentifier::named`). Id-keyed effects
//! return once reply-body parsing lands.
diff --git a/core/simulator/src/workload/shadow.rs b/core/simulator/src/workload/shadow.rs
index 2692ff168..b0e0ea8cb 100644
--- a/core/simulator/src/workload/shadow.rs
+++ b/core/simulator/src/workload/shadow.rs
@@ -17,7 +17,7 @@
//! Shadow state: the workload's prediction of server-side entity state.
//!
-//! Name-keyed throughout. Server-ng does not yet ship reply bodies, so
+//! Name-keyed throughout. Server does not yet ship reply bodies, so
//! the workload cannot observe server-assigned numeric ids. Lookups are
//! by name; requests route via `WireIdentifier::named(...)`. When typed
//! response bodies land, id-keyed maps return as a parallel index;
diff --git a/examples/node/package-lock.json b/examples/node/package-lock.json
index fdf8f7883..2847ff873 100644
--- a/examples/node/package-lock.json
+++ b/examples/node/package-lock.json
@@ -23,7 +23,7 @@
},
"../../foreign/node": {
"name": "apache-iggy",
- "version": "0.10.0-edge.5",
+ "version": "0.10.0-edge.6",
"license": "Apache-2.0",
"dependencies": {
"@node-rs/xxhash": "1.7.7",
diff --git a/examples/python/uv.lock b/examples/python/uv.lock
index bde592d6b..469565c7e 100644
--- a/examples/python/uv.lock
+++ b/examples/python/uv.lock
@@ -8,7 +8,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev6"
+version = "0.9.0.dev7"
source = { directory = "../../foreign/python" }
[package.metadata]
diff --git a/foreign/go/contracts/version.go b/foreign/go/contracts/version.go
index 32ff7a282..f3799086d 100644
--- a/foreign/go/contracts/version.go
+++ b/foreign/go/contracts/version.go
@@ -17,4 +17,4 @@
package iggcon
-const Version = "0.9.0-edge.5"
+const Version = "0.9.0-edge.6"
diff --git a/foreign/go/errors/errors.yaml b/foreign/go/errors/errors.yaml
index 7219af03e..e9f8b7c96 100644
--- a/foreign/go/errors/errors.yaml
+++ b/foreign/go/errors/errors.yaml
@@ -777,7 +777,7 @@
type: string
- name: TooManyConsumerOffsets
code: 3024
- format: "too many consumer offsets"
+ format: "too many consumer offsets for partition"
fields: []
- name: PartitionIdSpaceExhausted
code: 3013
diff --git a/foreign/go/errors/errors_gen.go b/foreign/go/errors/errors_gen.go
index 67641e984..06031cd66 100644
--- a/foreign/go/errors/errors_gen.go
+++ b/foreign/go/errors/errors_gen.go
@@ -1586,7 +1586,7 @@ func (e CannotOpenConsumerOffsetsFile) Is(target error) bool {
type TooManyConsumerOffsets struct{}
-func (e TooManyConsumerOffsets) Error() string { return "too many consumer offsets" }
+func (e TooManyConsumerOffsets) Error() string { return "too many consumer offsets for partition" }
func (e TooManyConsumerOffsets) Code() Code { return 3024 }
func (e TooManyConsumerOffsets) Is(target error) bool {
_, ok := target.(TooManyConsumerOffsets)
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java
index 992f3b094..5d776f827 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java
@@ -93,6 +93,8 @@ public enum IggyErrorCode {
// Partition errors
PARTITION_NOT_FOUND(3007),
PARTITION_ID_SPACE_EXHAUSTED(3013),
+
+ // Consumer offset errors
TOO_MANY_CONSUMER_OFFSETS(3024),
// Segment errors
diff --git a/foreign/node/package-lock.json b/foreign/node/package-lock.json
index 69b394e60..10bbf5baf 100644
--- a/foreign/node/package-lock.json
+++ b/foreign/node/package-lock.json
@@ -1,12 +1,12 @@
{
"name": "apache-iggy",
- "version": "0.10.0-edge.5",
+ "version": "0.10.0-edge.6",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "apache-iggy",
- "version": "0.10.0-edge.5",
+ "version": "0.10.0-edge.6",
"license": "Apache-2.0",
"dependencies": {
"@node-rs/xxhash": "1.7.7",
diff --git a/foreign/node/package.json b/foreign/node/package.json
index 6722c7161..ea4ec4dd0 100644
--- a/foreign/node/package.json
+++ b/foreign/node/package.json
@@ -1,7 +1,7 @@
{
"name": "apache-iggy",
"type": "module",
- "version": "0.10.0-edge.5",
+ "version": "0.10.0-edge.6",
"description": "Official Apache Iggy NodeJS SDK",
"keywords": [
"iggy",
diff --git a/foreign/node/src/wire/error.code.test.ts b/foreign/node/src/wire/error.code.test.ts
index fc4cb34cb..edbaf7367 100644
--- a/foreign/node/src/wire/error.code.test.ts
+++ b/foreign/node/src/wire/error.code.test.ts
@@ -20,7 +20,7 @@ import { it } from 'node:test';
import { translateErrorCode } from './error.code.js';
it('translates the consumer-offset capacity error', () => {
- assert.equal(translateErrorCode(3024), 'Too many consumer offsets');
+ assert.equal(translateErrorCode(3024), 'Too many consumer offsets for partition');
});
it('translates the consumer-group error range', () => {
diff --git a/foreign/node/src/wire/error.code.ts b/foreign/node/src/wire/error.code.ts
index d0a1f3d78..2cdb31e2f 100644
--- a/foreign/node/src/wire/error.code.ts
+++ b/foreign/node/src/wire/error.code.ts
@@ -165,7 +165,7 @@ export const translateErrorCode = (code: number): string => {
case '3021': return "Consumer offset for consumer with ID: {0} was not found.";
case '3022': return "Failed to resolve consumer with ID: {0}";
case '3023': return "Cannot open consumer offsets file for path: {0}";
- case '3024': return "Too many consumer offsets";
+ case '3024': return "Too many consumer offsets for partition";
case '3013': return "Partition id space exhausted for this topic";
// MESSAGE
diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml
index f3965a16e..1430a98fc 100644
--- a/foreign/python/Cargo.toml
+++ b/foreign/python/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "apache-iggy"
-version = "0.9.0-dev6"
+version = "0.9.0-dev7"
edition = "2024"
authors = ["Iggy Committers <de...@iggy.apache.org>"]
license = "Apache-2.0"
@@ -37,7 +37,7 @@ doc = false
[dependencies]
bytes = "1.12.1"
futures = "0.3.34"
-iggy = { path = "../../core/sdk", version = "0.11.0-edge.6" }
+iggy = { path = "../../core/sdk", version = "0.11.0-edge.7" }
paste = "1"
pyo3 = "0.29.2"
pyo3-async-runtimes = { version = "0.29.0", features = [
diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml
index 54d53ec90..1434010dc 100644
--- a/foreign/python/pyproject.toml
+++ b/foreign/python/pyproject.toml
@@ -22,7 +22,7 @@ build-backend = "maturin"
[project]
name = "apache-iggy"
requires-python = ">=3.10"
-version = "0.9.0.dev6"
+version = "0.9.0.dev7"
description = "Apache Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
readme = "README.md"
license = { file = "LICENSE" }
diff --git a/foreign/python/uv.lock b/foreign/python/uv.lock
index d6f3eb506..f4fce79c9 100644
--- a/foreign/python/uv.lock
+++ b/foreign/python/uv.lock
@@ -8,7 +8,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev6"
+version = "0.9.0.dev7"
source = { editable = "." }
[package.optional-dependencies]