krishvishal commented on code in PR #4392:
URL: https://github.com/apache/iggy/pull/4392#discussion_r4188047947
##########
core/partitions/src/state_transfer.rs:
##########
@@ -64,7 +64,11 @@ use std::sync::atomic::Ordering;
/// Current state-transfer offsets format, including the prepare-chain anchor.
pub(crate) const CONSUMER_OFFSETS_MAGIC: [u8; 4] = *b"ICO1";
-pub(crate) const CONSUMER_OFFSETS_VERSION: u8 = 1;
+/// Version 2: the external group count and section. `server-0.9.0` sends
+/// version 1, which has neither, so an artifact exchanged with a peer on that
+/// release fails with `UnsupportedVersion` instead of decoding 4 bytes out of
+/// place.
+pub(crate) const CONSUMER_OFFSETS_VERSION: u8 = 2;
Review Comment:
Fixed: decode accepts versions 1 and 2, and encode writes version 1 when
there is no external offset. A test checks the exact `server-0.9.0` bytes in
both directions.
##########
core/server/src/namespace.rs:
##########
@@ -251,11 +253,17 @@ where
S: 'static,
SB: SuperblockStore + 'static,
{
- if consumer.kind != KIND_CONSUMER_GROUP {
- return Ok(());
+ let streams = shard.plane.metadata().mux_stm.streams();
+ match consumer.kind {
+ KIND_CONSUMER_GROUP => {}
+ KIND_EXTERNAL_GROUP => {
Review Comment:
Partly: added the upgrade note to `core/server/README.md`. A cluster-wide
gate needs each replica's release, and no header carries it yet.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2092,17 +2146,14 @@ where
.retain(|start_offset, _| live.contains_key(start_offset));
}
- // An empty chain at frontier 0 tells the receiver to unlink its own,
so
- // serve it only when a recorded purge says the emptiness is the truth.
- // `install_state_transfer`'s `purge_advances` check re-decides that
- // against the metadata plane and refuses the rest.
+ // An empty chain at frontier 0 is still served. The `NothingCommitted`
Review Comment:
Fixed with both guards: hubcio's gate stays, and the install also refuses an
offer that holds nothing while the receiver holds committed offsets, with the
test you asked for.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -8022,38 +7981,42 @@ where
}
// Directory fsync so those unlinks stick, mirroring the install path:
a
// crash right after the purge otherwise resurrects the offset files at
- // boot, and while recovery clamps a resurrected offset down to the
- // rebuilt head, "consumed through 0" is not the intended "no entry at
- // all" -- that consumer skips the first post-purge message. Logged on
- // failure, sharper than the partition-dir fsync above: the generation
- // write below still runs, so a crash would resurrect these files with
- // no boot re-purge left to clear them.
- for dir in self
- .consumer_offsets_path
- .clone()
- .into_iter()
- .chain(self.consumer_group_offsets_path.clone())
- {
- if let Err(error) = storage.sync_directory(Path::new(&dir)).await {
- warn!(
- target: "iggy.partitions.diag",
- plane = "partitions",
- namespace_raw = namespace.inner(),
- generation,
- dir = %dir,
- %error,
- "purge could not fsync an offsets dir; a crash may
resurrect \
- deleted offset files with the purge already recorded"
- );
+ // boot. Recovery clamps a resurrected Iggy offset down to the rebuilt
+ // head, but "consumed through 0" is not the intended "no entry at all"
+ // -- that consumer skips the first post-purge message -- and boot
never
+ // clamps an external group offset, so one of 1000 skips 1000 messages.
+ // A failed sync therefore leaves the generation unrecorded below, and
+ // the reconciler re-issues the purge.
+ let mut offset_dirs_synced = true;
+ for dir in self.consumer_offset_dirs.iter().flatten() {
+ match storage.sync_directory(Path::new(dir)).await {
+ Ok(()) => {}
+ Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+ // Recovery accepts a missing directory, and a directory
+ // that is gone holds no unlink to make durable.
+ }
+ Err(error) => {
Review Comment:
Fixed: a dir sync failure other than `NotFound` now returns
`OffsetsNotDurable`, and the router fences the partition.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -8099,6 +8062,11 @@ where
.len();
self.evict_committed_prefix(fenced_prefix).await;
+ if !offset_dirs_synced {
Review Comment:
Fixed in round 2. Now only a failed `purge.gen` write is retried, and the
retry redoes only that write.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
