This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch consumer_offset_clamp in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 8374ca7ad600f49ac4a4259b44f8d766efa18e08 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Tue Nov 3 16:37:22 2026 +0100 fix(server): clamp recovered consumer offsets after the WAL replay --- 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 130f544a8..8d91b2399 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13070,7 +13070,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
