This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 8f2c9ffd6 fix(server): clamp recovered consumer offsets after the WAL
replay (#4737)
8f2c9ffd6 is described below
commit 8f2c9ffd65e21ff04dba02f7cbd4cfd12adc62a5
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Tue Nov 3 22:16:13 2026 +0100
fix(server): clamp recovered consumer offsets after the WAL replay (#4737)
Closes #4726
---
Cargo.lock | 2 +-
.../tests/server/consumer_offset_recovery.rs | 142 ++++++++++++++++
core/integration/tests/server/mod.rs | 1 +
core/partitions/src/iggy_partition.rs | 18 +++
core/partitions/src/lib.rs | 9 +-
core/partitions/src/partition_storage.rs | 180 +++++++++++++--------
core/server/Cargo.toml | 2 +-
core/server/src/partition_helpers.rs | 19 ++-
8 files changed, 296 insertions(+), 77 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 38cd9f10a..efbb11ce1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -13072,7 +13072,7 @@ dependencies = [
[[package]]
name = "server"
-version = "0.9.1-edge.8"
+version = "0.9.1-edge.9"
dependencies = [
"ahash",
"arc-swap",
diff --git a/core/integration/tests/server/consumer_offset_recovery.rs
b/core/integration/tests/server/consumer_offset_recovery.rs
new file mode 100644
index 000000000..f799602f0
--- /dev/null
+++ b/core/integration/tests/server/consumer_offset_recovery.rs
@@ -0,0 +1,142 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! A crash restart must not walk a stored consumer offset back below the data
that the WAL
+//! replay restores (apache/iggy#4726).
+
+use iggy::prelude::*;
+use iggy_common::ConsumerOffsetInfo;
+use integration::harness::{TestBinaryError, TestHarness};
+use integration::iggy_harness;
+use std::time::Duration;
+use test_case::test_matrix;
+use tokio::time::{Instant, sleep};
+
+const STREAM_NAME: &str = "consumer-offset-recovery-stream";
+const TOPIC_NAME: &str = "consumer-offset-recovery-topic";
+const PARTITION_ID: u32 = 0;
+/// Below the default `messages_required_to_save`, so no segment or WAL
checkpoint names any of
+/// them before the crash, and only the WAL replay restores them.
+const MESSAGES_COUNT: u64 = 500;
+const BATCH_SIZE: u64 = 100;
+const RECOVERY_TIMEOUT: Duration = Duration::from_secs(60);
+const RETRY_INTERVAL: Duration = Duration::from_millis(100);
+
+/// With replicated offsets the replay may not restore the offset store, so an
early clamp would
+/// stick. With persisted offsets it does, and only the clamp's warning would
show.
+#[iggy_harness(cluster_nodes = 1)]
+#[test_matrix([replicated_offsets(), persisted_offsets()])]
+async fn
given_persisted_messages_when_the_server_crashes_should_keep_the_stored_offset(
+ harness: &mut TestHarness,
+ consumer_offset_durability: Durability,
+) {
+ let stream = Identifier::named(STREAM_NAME).expect("stream identifier");
+ let topic = Identifier::named(TOPIC_NAME).expect("topic identifier");
+ let consumer = Consumer::new(Identifier::numeric(1).expect("consumer
identifier"));
+ let last_offset = MESSAGES_COUNT - 1;
+
+ let client = harness.tcp_root_client().await.expect("TCP root client");
+ client
+ .create_stream(STREAM_NAME, &StreamCreateOptions::default())
+ .await
+ .expect("create stream");
+ client
+ .create_topic(
+ &stream,
+ TOPIC_NAME,
+ &TopicCreateOptions {
+ partitions_count: Some(1),
+ durability: Durability::Persisted,
+ consumer_offset_durability,
+ ..TopicCreateOptions::default()
+ },
+ )
+ .await
+ .expect("create topic");
+ for batch in 0..MESSAGES_COUNT / BATCH_SIZE {
+ let mut messages = (0..BATCH_SIZE)
+ .map(|index| {
+ IggyMessage::builder()
+ .payload(format!("message-{}", batch * BATCH_SIZE +
index).into())
+ .build()
+ .expect("build message")
+ })
+ .collect::<Vec<_>>();
+ client
+ .send_messages(
+ &stream,
+ &topic,
+ &Partitioning::partition_id(PARTITION_ID),
+ &mut messages,
+ )
+ .await
+ .expect("send messages");
+ }
+ client
+ .store_consumer_offset(&consumer, &stream, &topic, Some(PARTITION_ID),
last_offset)
+ .await
+ .expect("store the offset of the last message");
+ drop(client);
+
+ harness.kill_node(0).expect("kill the server");
+ harness.restart_node(0).expect("restart the server");
+
+ let recovered = wait_for_recovered_offset(harness, &consumer, &stream,
&topic, last_offset)
+ .await
+ .expect("the restarted server should serve the consumer offset");
+ assert_eq!(
+ recovered.stored_offset, last_offset,
+ "the crash must not walk the stored offset back below the recovered
data"
+ );
+}
+
+fn replicated_offsets() -> Durability {
+ Durability::Replicated
+}
+
+fn persisted_offsets() -> Durability {
+ Durability::Persisted
+}
+
+/// Waits until the restarted server answers with the partition data restored
up to `last_offset`.
+async fn wait_for_recovered_offset(
+ harness: &TestHarness,
+ consumer: &Consumer,
+ stream: &Identifier,
+ topic: &Identifier,
+ last_offset: u64,
+) -> Result<ConsumerOffsetInfo, TestBinaryError> {
+ let deadline = Instant::now() + RECOVERY_TIMEOUT;
+ loop {
+ if let Ok(client) = harness.tcp_root_client().await
+ && let Ok(Some(offset)) = client
+ .get_consumer_offset(consumer, stream, topic,
Some(PARTITION_ID))
+ .await
+ && offset.current_offset == last_offset
+ {
+ return Ok(offset);
+ }
+ if Instant::now() >= deadline {
+ return Err(TestBinaryError::InvalidState {
+ message: format!(
+ "the partition did not recover up to offset {last_offset}
within {RECOVERY_TIMEOUT:?}"
+ ),
+ });
+ }
+ sleep(RETRY_INTERVAL).await;
+ }
+}
diff --git a/core/integration/tests/server/mod.rs
b/core/integration/tests/server/mod.rs
index a5e16b8cf..001fac308 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -81,6 +81,7 @@ mod concurrent_addition;
// The node-wide cap on client sockets closes a socket past it at accept.
mod connections_limit_vsr;
mod consumer_offset_quota_vsr;
+mod consumer_offset_recovery;
// A write that runs out of file descriptors flushes, then exits with status 4.
mod descriptor_exhaustion_vsr;
mod general;
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 61705d6dc..3b3f0fe9b 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -3915,6 +3915,24 @@ where
);
}
+ /// Lower the logical value of a recovered offset that lies past the
partition's offset space,
+ /// keeping the high-water read from its file.
+ pub fn clamp_recovered_consumer_offset(
+ &self,
+ kind: ConsumerKind,
+ consumer_id: u32,
+ committed_offset: u64,
+ ) {
+ if let Some(state) = self.durable_consumer_offsets.get(kind,
consumer_id) {
+ self.durable_consumer_offsets.record_explicit(
+ kind,
+ consumer_id,
+ committed_offset,
+ state.persisted_high_water,
+ );
+ }
+ }
+
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)
diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs
index 0d0066ce3..67247a376 100644
--- a/core/partitions/src/lib.rs
+++ b/core/partitions/src/lib.rs
@@ -102,10 +102,11 @@ pub use iggy_partition::{PollCompletion, PollReplication,
PollVisibility};
pub use messages_writer::MessagesWriter;
pub use offset_storage::delete_persisted_offset;
pub use partition_storage::{
- CREATED_REVISION_FILE, configure_consumer_offsets,
configure_consumer_offsets_with_storage,
- create_partition_file_hierarchy, delete_partitions_from_disk,
ensure_initial_segment,
- hydrate_partition_log, read_created_revision, read_revision_record,
write_created_revision,
- write_revision_record,
+ CREATED_REVISION_FILE, clamp_consumer_offsets, configure_consumer_offsets,
+ configure_consumer_offsets_with_storage, create_partition_file_hierarchy,
+ delete_partitions_from_disk, ensure_initial_segment, hydrate_partition_log,
+ read_created_revision, read_revision_record, recover_consumer_offsets,
+ recover_consumer_offsets_with_storage, write_created_revision,
write_revision_record,
};
pub use poll_plan::{DISK_POLL_CHUNK_MAX, PollPlan, PollReadResult};
pub use segment::Segment;
diff --git a/core/partitions/src/partition_storage.rs
b/core/partitions/src/partition_storage.rs
index a24415b11..2daaa586a 100644
--- a/core/partitions/src/partition_storage.rs
+++ b/core/partitions/src/partition_storage.rs
@@ -209,6 +209,20 @@ pub async fn configure_consumer_offsets<B: MessageBus>(
.await
}
+/// Populate `partition` with consumer and consumer group offset storage from
disk without
+/// clamping the recovered offsets, see
[`recover_consumer_offsets_with_storage`].
+///
+/// # Errors
+///
+/// The errors of [`recover_consumer_offsets_with_storage`].
+pub async fn recover_consumer_offsets<B: MessageBus>(
+ partition: &mut IggyPartition<B>,
+ config: &PartitionsConfig,
+ namespace: IggyNamespace,
+) -> Result<(), PartitionRecoveryError> {
+ recover_consumer_offsets_with_storage(&DiskStorage, partition, config,
namespace).await
+}
+
/// Recover consumer and group offsets from `storage` into a new partition.
///
/// Restore the partition's message offset and reservation frontier before
@@ -231,13 +245,34 @@ pub async fn configure_consumer_offsets<B: MessageBus>(
/// unreadable or stale emptied-table marker cannot be removed. Consumer
recovery may
/// already have seeded the partition when group recovery fails, so callers
must
/// discard a failed recovery.
-#[allow(clippy::too_many_lines)]
pub async fn configure_consumer_offsets_with_storage<S: DurableStorage, B:
MessageBus>(
storage: &S,
partition: &mut IggyPartition<B>,
config: &PartitionsConfig,
namespace: IggyNamespace,
current_offset: u64,
+) -> Result<(), PartitionRecoveryError> {
+ recover_consumer_offsets_with_storage(storage, partition, config,
namespace).await?;
+ clamp_consumer_offsets(partition, namespace, current_offset);
+ Ok(())
+}
+
+/// Recover consumer and group offsets from `storage` into a new partition, at
the values read
+/// from storage.
+///
+/// For a boot that replays its WAL afterwards: before the replay, neither the
replayed data nor
+/// the replayed offset stores bound the recovered offsets. Call
[`clamp_consumer_offsets`] once
+/// the replay is done.
+///
+/// # Errors
+/// The errors of [`configure_consumer_offsets_with_storage`]. Consumer
recovery may already have
+/// seeded the partition when group recovery fails, so callers must discard a
failed recovery.
+#[allow(clippy::too_many_lines)]
+pub async fn recover_consumer_offsets_with_storage<S: DurableStorage, B:
MessageBus>(
+ storage: &S,
+ partition: &mut IggyPartition<B>,
+ config: &PartitionsConfig,
+ namespace: IggyNamespace,
) -> Result<(), PartitionRecoveryError> {
let stream_id = namespace.stream_id();
let topic_id = namespace.topic_id();
@@ -247,17 +282,6 @@ pub async fn configure_consumer_offsets_with_storage<S:
DurableStorage, B: Messa
config.get_consumer_group_offsets_path(stream_id, topic_id,
partition_id);
let external_group_offsets_path =
config.get_external_group_offsets_path(stream_id, topic_id,
partition_id);
- // The bound is the offset space this replica could have MINTED, not the
data
- // it can still serve. A boot re-anchor leaves the append point a lease
block
- // above the recovered chain, so on the restart after a crash that took
- // acked-but-unflushed messages, a position stored before that crash names
a
- // real offset sitting under an empty chain -- confirmed to a client, and
not
- // "past the log" the way a torn offset file is. Bounding it by the data
head
- // instead walks a committed consumer position BACKWARD across the restart,
- // which is the silent re-read the reservation exists to prevent.
- // `mint_frontier` is one past the next mint, and reads 0 on the
fresh-build
- // path, where the max leaves `current_offset` in charge as before.
- let offset_space_ceiling =
current_offset.max(partition.mint_frontier().saturating_sub(1));
let recovered_consumers = load_partition_consumer_offsets(
storage,
@@ -272,14 +296,7 @@ pub async fn configure_consumer_offsets_with_storage<S:
DurableStorage, B: Messa
{
let guard = consumer_offsets.pin();
for offset in recovered_consumers.entries {
- seed_recovered_offset(
- partition,
- namespace,
- ConsumerKind::Consumer,
- &offset,
- Some(offset_space_ceiling),
- current_offset,
- );
+ seed_recovered_offset(partition, ConsumerKind::Consumer, &offset);
guard.insert(offset.consumer_id as usize, offset);
}
}
@@ -298,14 +315,7 @@ pub async fn configure_consumer_offsets_with_storage<S:
DurableStorage, B: Messa
{
let guard = consumer_group_offsets.pin();
for (group_id, offset) in recovered_groups.entries {
- seed_recovered_offset(
- partition,
- namespace,
- ConsumerKind::ConsumerGroup,
- &offset,
- Some(offset_space_ceiling),
- current_offset,
- );
+ seed_recovered_offset(partition, ConsumerKind::ConsumerGroup,
&offset);
guard.insert(group_id, offset);
}
}
@@ -334,14 +344,7 @@ pub async fn configure_consumer_offsets_with_storage<S:
DurableStorage, B: Messa
// Never clamped: the value belongs to the external system, and a Kafka
commit sits one past
// the last message. The durable table is the only store of an external
offset.
for (_, offset) in &recovered_external.entries {
- seed_recovered_offset(
- partition,
- namespace,
- ConsumerKind::ExternalGroup,
- offset,
- None,
- current_offset,
- );
+ seed_recovered_offset(partition, ConsumerKind::ExternalGroup, offset);
}
// Offset files have their own knob, not the topic's `persisted`: that
@@ -424,44 +427,95 @@ async fn restore_emptied_marker<S: DurableStorage, B:
MessageBus>(
Ok(())
}
-/// Seed the durable state of one recovered offset record. A record past
-/// `ceiling` is clamped to `current_offset` first, and `None` skips the clamp.
-/// The file high-water keeps the value read from storage.
+/// Seed the durable state of one recovered offset record at the value read
from storage.
fn seed_recovered_offset<B: MessageBus>(
partition: &IggyPartition<B>,
- namespace: IggyNamespace,
kind: ConsumerKind,
offset: &ConsumerOffset,
- ceiling: Option<u64>,
- current_offset: u64,
) {
let recovered_offset = offset.offset.load(Ordering::Relaxed);
- if let Some(ceiling) = ceiling
- && recovered_offset > ceiling
- {
- // A crash can persist an offset ahead of the flushed data (offsets are
- // stored eagerly, messages flush later). Clamp to the recovered head
so
- // the consumer resumes instead of being stuck polling past the log;
- // mirrors the legacy contract.
- warn!(
- stream_id = namespace.stream_id(),
- topic_id = namespace.topic_id(),
- partition_id = namespace.partition_id(),
- %kind,
- consumer_id = offset.consumer_id,
- recovered_offset,
- current_offset,
- offset_space_ceiling = ceiling,
- "recovered offset ahead of partition data; clamping"
- );
- offset.offset.store(current_offset, Ordering::Relaxed);
- }
partition.seed_recovered_consumer_offset(
kind,
offset.consumer_id,
- offset.offset.load(Ordering::Relaxed),
recovered_offset,
+ recovered_offset,
+ );
+}
+
+/// Clamp the consumer and group offsets past the offset space `partition`
could have minted.
+///
+/// A clamped offset moves to `current_offset`. Clamping changes the visible
position without
+/// rewriting its file, and persistence tracking keeps the value read from
storage. External
+/// group offsets are never clamped: the value belongs to the external system.
+pub fn clamp_consumer_offsets<B: MessageBus>(
+ partition: &IggyPartition<B>,
+ namespace: IggyNamespace,
+ current_offset: u64,
+) {
+ // The bound is the offset space this replica could have MINTED, not the
data
+ // it can still serve. A boot re-anchor leaves the append point a lease
block
+ // above the recovered chain, so on the restart after a crash that took
+ // acked-but-unflushed messages, a position stored before that crash names
a
+ // real offset sitting under an empty chain -- confirmed to a client, and
not
+ // "past the log" the way a torn offset file is. Bounding it by the data
head
+ // instead walks a committed consumer position BACKWARD across the restart,
+ // which is the silent re-read the reservation exists to prevent.
+ // `mint_frontier` is one past the next mint, and reads 0 on the
fresh-build
+ // path, where the max leaves `current_offset` in charge as before.
+ let ceiling =
current_offset.max(partition.mint_frontier().saturating_sub(1));
+ let consumers = partition.consumer_offsets.pin();
+ for (_, offset) in &consumers {
+ clamp_recovered_offset(
+ partition,
+ namespace,
+ ConsumerKind::Consumer,
+ offset,
+ ceiling,
+ current_offset,
+ );
+ }
+ let groups = partition.consumer_group_offsets.pin();
+ for (_, offset) in &groups {
+ clamp_recovered_offset(
+ partition,
+ namespace,
+ ConsumerKind::ConsumerGroup,
+ offset,
+ ceiling,
+ current_offset,
+ );
+ }
+}
+
+fn clamp_recovered_offset<B: MessageBus>(
+ partition: &IggyPartition<B>,
+ namespace: IggyNamespace,
+ kind: ConsumerKind,
+ offset: &ConsumerOffset,
+ ceiling: u64,
+ current_offset: u64,
+) {
+ let recovered_offset = offset.offset.load(Ordering::Relaxed);
+ if recovered_offset <= ceiling {
+ return;
+ }
+ // A crash can persist an offset ahead of the flushed data (offsets are
+ // stored eagerly, messages flush later). Clamp to the recovered head so
+ // the consumer resumes instead of being stuck polling past the log;
+ // mirrors the legacy contract.
+ warn!(
+ stream_id = namespace.stream_id(),
+ topic_id = namespace.topic_id(),
+ partition_id = namespace.partition_id(),
+ %kind,
+ consumer_id = offset.consumer_id,
+ recovered_offset,
+ current_offset,
+ offset_space_ceiling = ceiling,
+ "recovered offset ahead of partition data; clamping"
);
+ offset.offset.store(current_offset, Ordering::Relaxed);
+ partition.clamp_recovered_consumer_offset(kind, offset.consumer_id,
current_offset);
}
/// Provision an initial segment + writers for a partition that has none.
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index 9e6f8f962..b2d3f457b 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "server"
-version = "0.9.1-edge.8"
+version = "0.9.1-edge.9"
edition = "2024"
license = "Apache-2.0"
publish = false
diff --git a/core/server/src/partition_helpers.rs
b/core/server/src/partition_helpers.rs
index 8aa95d568..a96443c3a 100644
--- a/core/server/src/partition_helpers.rs
+++ b/core/server/src/partition_helpers.rs
@@ -23,8 +23,8 @@
//! directory that belongs to another incarnation;
//! [`build_partition_fresh`] materialises one for a namespace that has no
//! directory yet. Both sit on the same namespace-bounds validation, and on
-//! the consumer-offset configuration and initial-segment provisioning in
-//! [`partitions::configure_consumer_offsets`] and
+//! the consumer-offset recovery and initial-segment provisioning in
+//! [`partitions::configure_consumer_offsets`],
[`partitions::recover_consumer_offsets`] and
//! [`partitions::ensure_initial_segment`].
//!
//! Boot runs the loader over every owned namespace. The reconciler picks
@@ -49,9 +49,10 @@ use metadata::stm::stream::Partition;
use metadata::{IdentityField, ReplicaIdentity};
use partitions::{
IggyPartition, IggyPartitions, PartitionPersistence,
PartitionRecoveryError,
- PartitionRecoveryRefusal, PartitionsConfig, RecoveredSegment,
configure_consumer_offsets,
- create_partition_file_hierarchy, ensure_initial_segment,
hydrate_partition_log,
- load_persisted_segments, load_persisted_segments_with_checkpoint,
+ PartitionRecoveryRefusal, PartitionsConfig, RecoveredSegment,
clamp_consumer_offsets,
+ configure_consumer_offsets, create_partition_file_hierarchy,
ensure_initial_segment,
+ hydrate_partition_log, load_persisted_segments,
load_persisted_segments_with_checkpoint,
+ recover_consumer_offsets,
};
use server_common::fatal::is_descriptor_exhaustion;
use server_common::fs_utils::run_blocking;
@@ -791,10 +792,8 @@ async fn load_partition(
partition.created_at = partition_metadata.created_at;
restore_partition_offsets(&mut partition, recovered_state.as_ref(),
segment_checkpoint);
- let current_offset = partition.offset.load(Ordering::Acquire);
- configure_consumer_offsets(&mut partition, partitions_config, namespace,
current_offset)
- .await?;
+ recover_consumer_offsets(&mut partition, partitions_config,
namespace).await?;
if !partition.log.has_segments() {
ensure_initial_segment(
&mut partition,
@@ -813,6 +812,10 @@ async fn load_partition(
.await
.map_err(|error| ServerError::Iggy(Box::new(error)))?;
}
+ // Only now does the ceiling see the replayed data and the re-anchored
reservation. An offset
+ // whose store the replay does not restore would otherwise stay clamped
below its data.
+ let current_offset = partition.offset.load(Ordering::Acquire);
+ clamp_consumer_offsets(&partition, namespace, current_offset);
if let Err(error) =
record_partition_initialization(config, namespace,
partition_metadata.created_revision)
.await
