This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/durable-protocol-and-sessions in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 77d9f513af56d17373964cfadc04c332a804f541 Author: Hubert Gruszecki <[email protected]> AuthorDate: Tue Oct 6 11:03:00 2026 +0200 test(partitions): align offset fixtures with durable retry state External offset fixtures need committed retry capacity and the current directory layout after merging durable sessions. Use valid retry snapshots, preserve transfer and quota assertions, and reject legacy artifacts that lack session-qualified receipts. --- core/partitions/src/iggy_partition.rs | 33 +++++++++++++++++++++------------ core/partitions/src/state_transfer.rs | 20 ++++++++++---------- 2 files changed, 31 insertions(+), 22 deletions(-) diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 43287b2a9..9260bb383 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -11218,12 +11218,7 @@ mod tests { std::fs::create_dir_all(&directory).unwrap(); let mut partition = partition_at_view(0, 0); partition.set_partition_dir(directory.to_string_lossy().into_owned()); - let consumers = directory.join("offsets/consumers"); - let groups = directory.join("offsets/groups"); - std::fs::create_dir_all(&consumers).unwrap(); - std::fs::create_dir_all(&groups).unwrap(); - partition.consumer_offsets_path = Some(consumers.to_string_lossy().into_owned()); - partition.consumer_group_offsets_path = Some(groups.to_string_lossy().into_owned()); + set_offset_dirs_under(&mut partition, &directory); partition.runtime_options.durability = iggy_common::Durability::Persisted; partition.open_persistence().await.unwrap(); partition.log.retire_front().unwrap(); @@ -11293,6 +11288,7 @@ mod tests { next_offset: 2, consumers: Vec::new(), groups: Vec::new(), + external_groups: Vec::new(), dedup: partition.dedup.watermarks_sorted(), }; partition @@ -13763,6 +13759,7 @@ mod tests { .into_owned(), ); let wire = crate::state_transfer::ConsumerOffsetsWire { + dedup_capacity: partition.dedup().capacity(), purge_generation: 1, external_groups: vec![(8, 0), (9, 1 << 40)], ..Default::default() @@ -13989,6 +13986,10 @@ mod tests { origin.consensus().sequencer().set_sequence(1); origin.consensus().advance_commit_max(1); origin.consensus().advance_commit_min(1); + origin + .dedup_mut() + .commit_capacity(consensus::CLIENTS_TABLE_MAX) + .unwrap(); let offer = origin.state_transfer_offer(&repair_config()).await.unwrap(); let (mut receiver, _) = recording_partition_at(1, 3); receiver.set_partition_dir(receiver_directory.path().to_string_lossy().into_owned()); @@ -14033,7 +14034,10 @@ mod tests { let offset_file = directory.path().join("offsets/external_groups/7"); assert!(offset_file.exists()); - let nothing = crate::state_transfer::ConsumerOffsetsWire::default(); + let nothing = crate::state_transfer::ConsumerOffsetsWire { + dedup_capacity: receiver.dedup().capacity(), + ..Default::default() + }; let refused = receiver .install_state_transfer(&repair_config(), 5, Vec::new(), ¬hing.encode(), 0) .await; @@ -14054,7 +14058,7 @@ mod tests { let purged = crate::state_transfer::ConsumerOffsetsWire { purge_generation: 1, - ..crate::state_transfer::ConsumerOffsetsWire::default() + ..nothing }; receiver .install_state_transfer(&repair_config(), 5, Vec::new(), &purged.encode(), 1) @@ -15614,7 +15618,8 @@ mod tests { async fn given_full_offset_table_when_storing_new_id_should_reject_before_replication() { let (mut partition, sent_to_clients) = recording_partition(); let directory = tempfile::tempdir().unwrap(); - partition.consumer_offsets_path = Some(directory.path().to_string_lossy().into_owned()); + partition.consumer_offset_dirs[ConsumerKind::Consumer.index()] = + Some(directory.path().to_string_lossy().into_owned()); partition.set_consumer_offsets_max(1); partition.stats.increment_messages_count(1); @@ -15657,7 +15662,8 @@ mod tests { async fn given_full_offset_table_when_updating_existing_id_should_succeed() { let (mut partition, _) = recording_partition(); let directory = tempfile::tempdir().unwrap(); - partition.consumer_offsets_path = Some(directory.path().to_string_lossy().into_owned()); + partition.consumer_offset_dirs[ConsumerKind::Consumer.index()] = + Some(directory.path().to_string_lossy().into_owned()); partition.set_consumer_offsets_max(1); partition.stats.increment_messages_count(1); for request_id in 1..=2 { @@ -15688,7 +15694,7 @@ mod tests { let directory = tempfile::tempdir().unwrap(); let (mut partition, replies) = recording_partition(); partition.set_partition_dir(directory.path().to_string_lossy().into_owned()); - partition.consumer_offsets_path = Some( + partition.consumer_offset_dirs[ConsumerKind::Consumer.index()] = Some( directory .path() .join("consumers") @@ -18387,7 +18393,10 @@ mod tests { partition.unrecorded_purge_generation = Some(1); partition.purge_deferred = true; - let offer = crate::state_transfer::ConsumerOffsetsWire::default(); + let offer = crate::state_transfer::ConsumerOffsetsWire { + dedup_capacity: partition.dedup().capacity(), + ..Default::default() + }; partition .install_state_transfer(&repair_config(), 12, Vec::new(), &offer.encode(), 1) .await diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs index 268671639..daca7bc70 100644 --- a/core/partitions/src/state_transfer.rs +++ b/core/partitions/src/state_transfer.rs @@ -73,7 +73,9 @@ const RETRY_CHECKPOINT_PREFIX: &str = "receipts-"; const RETRY_CHECKPOINT_SUFFIX: &str = ".checkpoint"; const RETRY_CHECKPOINT_TEMP_SUFFIX: &str = ".checkpoint.tmp"; const RETRY_CHECKPOINT_BYTES_MAX: usize = journal::partition_journal::PREPARE_BYTES_MAX - + ConsumerKind::COUNT * CONSUMER_OFFSETS_ENTRIES_MAX as usize * (size_of::<u32>() + size_of::<u64>()) + + ConsumerKind::COUNT + * CONSUMER_OFFSETS_ENTRIES_MAX as usize + * (size_of::<u32>() + size_of::<u64>()) + CLIENTS_TABLE_SLOT_MAX * (DEDUP_ENTRY_LEN + PARTITION_RECEIPT_BYTES_MAX) + size_of::<PrepareHeader>(); @@ -975,18 +977,14 @@ mod tests { ); } assert_eq!( - CONSUMER_OFFSETS_VERSION, - 4, + CONSUMER_OFFSETS_VERSION, 4, "a consumer-offsets version moved; confirm its layout moved with it" ); } - /// `server-0.9.0` writes version 1 with this layout. Decoding it has to give - /// the same table, and a table with no external group offset has to encode - /// to the same bytes, or every partition transfer between the two releases - /// fails during a rolling upgrade. + /// Legacy artifacts omit session-qualified receipts and cannot restore retry protection. #[test] - fn given_server_0_9_0_artifact_when_decoded_and_encoded_again_should_match_it() { + fn given_server_0_9_0_artifact_when_decoded_should_reject() { let wire = ConsumerOffsetsWire { prepare_checksum: Some(u128::MAX - 7), checkpoint_prepare: vec![1, 2, 3], @@ -1022,8 +1020,10 @@ mod tests { let trailer = state_artifact_checksum(&released); released.extend_from_slice(&trailer.to_le_bytes()); - assert_eq!(ConsumerOffsetsWire::decode(&released), Ok(wire.clone())); - assert_eq!(wire.encode(), released); + assert_eq!( + ConsumerOffsetsWire::decode(&released), + Err(ConsumerOffsetsWireError::UnsupportedVersion { version: 1 }) + ); } #[test]
