You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/21 21:44:52 UTC
(iggy) 01/03: fix(metadata): place primaries on responsive replicas
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/spread-placement
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit dfc57cce2aa2e0e6e936a4e86054f4b374363ffd
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Wed Oct 21 23:42:34 2026 +0200
fix(metadata): place primaries on responsive replicas
Use readiness from the current metadata view for new assignments.
Missing observations and unavailable primaries no longer block
transfers of other healthy partitions.
---
core/metadata/src/impls/metadata.rs | 44 +++++++++++++++++++++--
core/metadata/src/impls/placement.rs | 67 +++++++++++++++++++++++++++---------
core/server/README.md | 10 +++---
core/shard/src/primary_rebalance.rs | 2 ++
4 files changed, 101 insertions(+), 22 deletions(-)
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index bf000b4a7..07bacbfdd 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -4271,6 +4271,16 @@ where
.observe(self.mux_stm.streams(), reports);
}
+ /// Initial placement uses readiness only within the metadata view that observed it.
+ /// An empty slice clears the evidence when the coordinator loses authority.
+ pub fn observe_replica_readiness(&self, eligible: &[bool]) {
+ if let Some(consensus) = self.consensus.as_ref() {
+ self.placement
+ .borrow_mut()
+ .observe_replicas(consensus.view(), eligible);
+ }
+ }
+
/// Suggest one count-balancing move only after normal failover has settled.
/// `after` rotates candidates so a recovering target cannot starve other groups.
pub fn primary_rebalance_candidate(
@@ -4676,7 +4686,10 @@ where
partition_id: u32::try_from(partition_id)
.map_err(|_| IggyError::InvalidCommand)?,
consensus_group_id,
- initial_view: PlacementState::select(&mut counts)?,
+ initial_view: self
+ .placement
+ .borrow()
+ .select(&mut counts, consensus.view())?,
})
})
.collect::<Result<Vec<_>, _>>()?;
@@ -4712,7 +4725,10 @@ where
partition_id: u32::try_from(offset)
.map_err(|_| IggyError::InvalidCommand)?,
consensus_group_id,
- initial_view: PlacementState::select(&mut counts)?,
+ initial_view: self
+ .placement
+ .borrow()
+ .select(&mut counts, consensus.view())?,
})
})
.collect::<Result<Vec<_>, _>>()?;
@@ -6290,6 +6306,30 @@ mod tests {
.collect()
}
+ #[test]
+ fn given_an_offline_replica_when_admitting_creates_should_spread_over_responsive_replicas() {
+ let metadata = placement_metadata();
+ metadata.observe_replica_readiness(&[false, true, true]);
+ let first = metadata
+ .prepare_request(placement_request(1, "first", 4))
+ .unwrap();
+ assert_eq!(assigned_views(&first), [1, 2, 1, 2]);
+ metadata
+ .consensus
+ .as_ref()
+ .unwrap()
+ .pipeline_message(PlaneKind::Metadata, &first);
+ let second = metadata
+ .prepare_request(placement_request(2, "second", 2))
+ .unwrap();
+ assert_eq!(assigned_views(&second), [1, 2]);
+ metadata.observe_replica_readiness(&[true; 3]);
+ let returned = metadata
+ .prepare_request(placement_request(3, "returned", 2))
+ .unwrap();
+ assert_eq!(assigned_views(&returned), [0, 0]);
+ }
+
#[test]
fn given_pending_placement_when_admitting_another_batch_should_spread_before_wal_append() {
let metadata = placement_metadata();
diff --git a/core/metadata/src/impls/placement.rs b/core/metadata/src/impls/placement.rs
index b2a029288..85ecfcdfd 100644
--- a/core/metadata/src/impls/placement.rs
+++ b/core/metadata/src/impls/placement.rs
@@ -32,6 +32,7 @@ use server_common::sharding::IggyNamespace;
pub(super) struct PlacementState {
observations: AHashMap<IggyNamespace, ObservedView>,
reservations: AHashMap<(u64, u128), Vec<u64>>,
+ eligible: Option<(u32, Vec<bool>)>,
}
struct ObservedView {
@@ -40,6 +41,10 @@ struct ObservedView {
}
impl PlacementState {
+ pub(super) fn observe_replicas(&mut self, view: u32, eligible: &[bool]) {
+ self.eligible = (!eligible.is_empty()).then(|| (view, eligible.to_vec()));
+ }
+
pub(super) fn observe(&mut self, streams: &Streams, reports: &[PartitionPlacementView]) {
let () = streams.read(|inner| {
for report in reports {
@@ -129,12 +134,14 @@ impl PlacementState {
}
for partition in &topic.partitions {
let namespace = IggyNamespace::new(stream_id, topic_id, partition.id);
- let report = self.matching_view(namespace, partition)?;
- let primary = report.view as usize % replicas;
- if !eligible[primary] {
- return None;
- }
+ let report = self.matching_view(namespace, partition);
+ let primary = report.map_or(partition.created_view, |report| report.view)
+ as usize
+ % replicas;
counts[primary] += 1;
+ let Some(report) = report.filter(|_| eligible[primary]) else {
+ continue;
+ };
if first[primary].is_none_or(|first| namespace < first.namespace) {
first[primary] = Some(report);
}
@@ -196,10 +203,14 @@ impl PlacementState {
})
}
- pub(super) fn select(counts: &mut [u64]) -> Result<u32, IggyError> {
+ pub(super) fn select(&self, counts: &mut [u64], view: u32) -> Result<u32, IggyError> {
+ let eligible = self.eligible.as_ref().filter(|(observed_view, eligible)| {
+ *observed_view == view && eligible.len() == counts.len()
+ });
let (replica, count) = counts
.iter_mut()
.enumerate()
+ .filter(|(replica, _)| eligible.is_none_or(|(_, eligible)| eligible[*replica]))
.min_by_key(|(replica, count)| (**count, *replica))
.ok_or(IggyError::TransientNotAccepted)?;
*count += 1;
@@ -435,11 +446,11 @@ mod tests {
},
] {
placement.observe(&streams, &[stale]);
- assert!(
- placement
- .primary_rebalance_candidate(&streams, 3, &[true; 3], None)
- .is_none()
- );
+ let transfer = placement
+ .primary_rebalance_candidate(&streams, 3, &[true; 3], None)
+ .unwrap();
+ assert_eq!(transfer.namespace, IggyNamespace::new(0, 1, 0));
+ assert_eq!(transfer.target, 0);
assert!(
placement
.partition_placement_view(&streams, namespace, report.created_revision)
@@ -538,11 +549,10 @@ mod tests {
..old_report
}],
);
- assert!(
- placement
- .primary_rebalance_candidate(&recreated, 3, &[true; 3], None)
- .is_none()
- );
+ let transfer = placement
+ .primary_rebalance_candidate(&recreated, 3, &[true; 3], None)
+ .unwrap();
+ assert_eq!(transfer.namespace, IggyNamespace::new(0, 1, 0));
assert!(
placement
.partition_placement_view(&recreated, namespace, new_revision)
@@ -567,4 +577,29 @@ mod tests {
.is_none()
);
}
+
+ #[test]
+ fn given_an_unavailable_primary_when_other_partitions_are_imbalanced_should_move_a_ready_one() {
+ let (streams, placement) = observed_topics(&[0, 1, 1, 1]);
+ let transfer = placement
+ .primary_rebalance_candidate(&streams, 3, &[false, true, true], None)
+ .unwrap();
+ assert_eq!(transfer.namespace, IggyNamespace::new(0, 1, 0));
+ assert_eq!(transfer.target, 2);
+ }
+
+ #[test]
+ fn given_readiness_from_another_view_when_placing_should_ignore_stale_evidence() {
+ let mut placement = PlacementState::default();
+ placement.observe_replicas(1, &[false, true, true]);
+ assert_eq!(placement.select(&mut [0, 5, 5], 1), Ok(1));
+ assert_eq!(placement.select(&mut [0, 5, 5], 2), Ok(0));
+ placement.observe_replicas(2, &[false; 3]);
+ assert_eq!(
+ placement.select(&mut [0, 5, 5], 2),
+ Err(IggyError::TransientNotAccepted)
+ );
+ placement.observe_replicas(2, &[]);
+ assert_eq!(placement.select(&mut [0, 5, 5], 2), Ok(0));
+ }
}
diff --git a/core/server/README.md b/core/server/README.md
index 46d08dbfe..6d0d4fdeb 100644
--- a/core/server/README.md
+++ b/core/server/README.md
@@ -120,13 +120,15 @@ Cluster membership, quorum and replica addressing live under `[cluster]`.
New partition primaries are spread across the configured replicas using each
partition's last observed primary, or its committed initial view before an
-observation arrives. Concurrent creates count pending assignments too. Every
-configured replica remains eligible, including an offline replica; ordinary
-partition failover handles that initial target. After the responsive replica set
+observation arrives. Concurrent creates count pending assignments too. Once
+readiness probes identify responsive replicas, new primaries are assigned only
+to those replicas. Before readiness is known, all configured replicas remain
+eligible. After the responsive replica set
has been stable for five seconds, the metadata primary balances existing primary
counts one partition at a time. It confirms the source's committed frontier and
the target's applied history before requesting a persisted consensus view change.
-Recovering partitions delay transfers. Every replica still stores every partition.
+Recovering partitions are skipped without delaying transfers of other partitions.
+Every replica still stores every partition.
Balancing primary counts does not balance bytes, throughput or CPU load. A hot
partition still has one ordering primary.
diff --git a/core/shard/src/primary_rebalance.rs b/core/shard/src/primary_rebalance.rs
index 439c2f449..e9787ff46 100644
--- a/core/shard/src/primary_rebalance.rs
+++ b/core/shard/src/primary_rebalance.rs
@@ -258,6 +258,7 @@ where
};
if self.id != 0 || !consensus.is_primary() || !consensus_ready(consensus) {
self.primary_rebalance.borrow_mut().reset_authority();
+ metadata.observe_replica_readiness(&[]);
return;
}
let mut outgoing = Vec::with_capacity(PROBES_PER_TICK);
@@ -316,6 +317,7 @@ where
if changed {
state.stable_since = now;
state.pending = None;
+ metadata.observe_replica_readiness(&state.eligible);
}
if let Some(pending) = state.pending.as_ref() {
let elapsed = now.duration_since(pending.since);