numinnex commented on code in PR #4392:
URL: https://github.com/apache/iggy/pull/4392#discussion_r4184460583
##########
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:
this drops the `segments.is_empty() && next_offset == 0 && purge_generation
== 0` gate from d35000d07, and the PR now makes offsets-only partitions
possible. A primary that lost its dir with `commit_max > 0` serves an empty
offer to a restarted replica (commit_min = commit_max = 0, so `StaleTransfer`
is inert), `OfferRewindsDurableData` only keys on `local_next_offset > 0`, and
the install sweep unlinks the receiver's committed external offsets.
Don't restore the blanket gate, extend the anti-rewind check to the durable
offset table, with a test where the receiver holds only external offsets. The
comment at :3398 still calls this shape the phantom the gate stops.
##########
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:
decode accepts only `== 2` and encode always writes v2, so every partition
state transfer between a 0.9.0 node and a new one fails with
`UnsupportedVersion` in both directions, even with zero external offsets. That
breaks the rolling upgrade `core/server/README.md` documents.
Accept `1 | 2` on decode (v1 = empty external section), write v1 when the
external section is empty, and add a v1 round-trip test.
##########
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:
`sync_directory` opens a fresh handle each time, so after a writeback EIO a
later fsync on the retried purge can succeed without the earlier unlinks being
durable (the comment at state_transfer.rs:3344 warns about exactly this).
`purge.gen` then gets recorded, and since boot never clamps external offsets a
crash can resurrect a deleted one.
On a non-NotFound error, fence the partition like `Unserviceable` instead of
relying on the retry.
##########
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:
a new primary admits kind 3 here, but a 0.9.0 backup fails
`WireConsumer::decode` on it before `journal_append`. No PrepareOk, repair
fails the same way, so with 2 of 3 replicas still old the partition stalls.
At minimum add an upgrade note to the README (upgrade the whole cluster
before any `ExternalGroup` store). A cluster-wide admission gate here would be
the real fix.
##########
core/sdk/src/http/consumer_offsets.rs:
##########
@@ -101,3 +104,12 @@ impl ConsumerOffsetClient for HttpClient {
fn get_path(stream_id: &str, topic_id: &str) -> String {
format!("streams/{stream_id}/topics/{topic_id}/consumer-offsets")
}
+
+/// The REST API cannot name a consumer kind, so an external group would land
on a plain
+/// consumer's offset.
+pub(crate) fn refuse_external_group(consumer: &Consumer) -> Result<(),
IggyError> {
Review Comment:
nit: nothing tests this. `Consumer::kind` is `#[serde(skip)]` on the REST
path, so if the guard goes, an external offset silently lands on a plain
consumer key. add a unit test.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -3301,38 +3264,39 @@ where
.collect()
}
- /// Snapshot dead group keys for deletion through the partition's VSR log.
- /// A local unlink could free the primary's quota while backups retained
- /// every older generation, so reclamation uses the same ordered delete as
- /// an explicit consumer-offset request.
+ /// Snapshot the committed offsets of every kind keyed by consumer group
+ /// whose group `is_live` rejects, ascending by id within each kind, for
+ /// deletion through the partition's VSR log. A local unlink could free the
+ /// primary's quota while backups retained every older generation, so
+ /// reclamation uses the same ordered delete as an explicit consumer-offset
+ /// request.
#[must_use]
- pub fn dead_consumer_group_offset_ids(&self, is_live: impl Fn(u64) ->
bool) -> Vec<u32> {
+ pub fn dead_group_offset_keys(
+ &self,
+ is_live: impl Fn(u64) -> bool,
+ ) -> Vec<(ConsumerKind, u32)> {
if !self.consensus.is_primary() || !self.consensus.is_normal() {
return Vec::new();
}
- // A stranded file already failed normal loading or unlink. Reissuing
- // replicated deletes every reconciliation pass cannot make its
- // filesystem writable and would create a permanent commit loop.
- // A single-replica explicit deletion can retry after repair. On a
- // replicated partition the file must be repaired or removed locally,
- // because older peers do not recognize a delete for a map-missing key.
- let capacity =
self.consumer_offset_capacity_for(ConsumerKind::ConsumerGroup);
let mut dead = Vec::new();
- for key in self.consumer_group_offsets.pin().keys() {
- let Ok(id) = u32::try_from(key.0) else {
- continue;
- };
- if !is_live(u64::from(id))
- && !capacity.is_stranded(id)
- && self
- .durable_consumer_offsets
- .contains(ConsumerKind::ConsumerGroup, id)
- {
- dead.push(id);
- }
+ for kind in [ConsumerKind::ConsumerGroup, ConsumerKind::ExternalGroup]
{
Review Comment:
nit: consumer-group keys go first under the 32-per-pass cap, so a big
backlog of dead CG keys delays external-group deletes by whole passes.
interleave, or note it is intended.
##########
core/partitions/src/partition_storage.rs:
##########
@@ -323,53 +283,92 @@ 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 {
- let recovered_offset = offset.offset.load(Ordering::Relaxed);
- if recovered_offset > offset_space_ceiling {
- warn!(
- consumer_group_id = group_id.0,
- recovered_offset,
- current_offset,
- offset_space_ceiling,
- stream_id,
- topic_id,
- partition_id,
- "recovered consumer group offset ahead of partition data;
clamping"
- );
- offset.offset.store(current_offset, Ordering::Relaxed);
- }
- let committed_offset = offset.offset.load(Ordering::Relaxed);
- partition.seed_recovered_consumer_offset(
+ seed_recovered_offset(
+ partition,
+ namespace,
ConsumerKind::ConsumerGroup,
- offset.consumer_id,
- committed_offset,
- recovered_offset,
+ &offset,
+ Some(offset_space_ceiling),
+ current_offset,
);
guard.insert(group_id, offset);
}
}
+ // A partition made before this kind existed has no directory for it. The
checkpoint syncs
+ // every offset directory, so it has to exist before the first one.
+ storage
+ .create_directories(Path::new(&external_group_offsets_path))
+ .await
+ .map_err(|_| PartitionRecoveryError::ConsumerOffsetsLoad {
Review Comment:
nit: `map_err(|_| ...)` drops the io error, so a failed create on the first
boot after upgrade says nothing about why. keep the source in the error.
##########
core/sdk/src/stream_builder/build/build_iggy_consumer.rs:
##########
@@ -58,6 +58,8 @@ pub(crate) async fn build_iggy_consumer(
let mut builder = match consumer_kind {
ConsumerKind::Consumer => client.consumer(consumer_name, stream,
topic, partition)?,
ConsumerKind::ConsumerGroup => client.consumer_group(consumer_name,
stream, topic)?,
+ // Holds offsets only, it cannot poll.
+ ConsumerKind::ExternalGroup => return
Err(IggyError::InvalidConfiguration),
Review Comment:
nit: the same misuse returns `InvalidConfiguration` here but
`FeatureUnavailable` from the server poll (dispatch/partition.rs:815) and the
HTTP SDK, and `FeatureUnavailable` now also means "wrong kind". pick one code
for it.
##########
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:
the re-issued purge wipes everything this replica committed since the first
attempt (messages, external stores). `purge_deferred` only gates the PrepareOk
send, local commits still apply, and peers keep those ops, so counters and the
next `base_offset` diverge here. Under persistent EIO every reconciler pass
wipes again with no backoff.
Retry only the dir syncs plus the `purge.gen` write (remember the wiped
generation in memory), or fence.
--
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]
