hubcio commented on code in PR #3374:
URL: https://github.com/apache/iggy/pull/3374#discussion_r3324795329
##########
core/server/src/metadata/consumer_group.rs:
##########
@@ -137,14 +137,21 @@ impl ConsumerGroupMeta {
return;
}
- // Find over-assigned members and mark excess as pending revocation
+ // Two-pass distribution: we first collect ALL excess partitions, then
distribute
+ // them round-robin. This ensures even distribution when one member
holds many
+ // partitions and multiple idle members join. Without this, the first
idle member
+ // would receive all excess partitions while others starve.
+ //
+ // Example: 16 partitions held by 1 member, 15 idle members join
+ // Single-pass: member1 gets 15, members 2-15 get 0-1 each
(unbalanced)
+ // Two-pass: each of 16 members gets exactly 1 partition
+
+ // Pass 1: Collect excess partitions from over-assigned members
let member_ids: Vec<usize> = self.members.iter().map(|(id, _)|
id).collect();
let mut members_with_remainder = remainder;
+ let mut all_excess: Vec<(PartitionId, usize)> = Vec::new();
for &mid in &member_ids {
- if idle_slab_ids.is_empty() {
- break;
- }
let effective_count = self
Review Comment:
the per-member `pending` set gets built twice for each over-assigned member
- once here in the `effective_count` closure, then again below in the
`revocable` closure (around line 186). same `HashSet<PartitionId>` from the
same `pending_revocations`. build it once before the `max_allowed` branch and
reuse for both, saves a few lines and one alloc per over-assigned member.
pre-existing, just surfaced by the restructure.
##########
core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs:
##########
@@ -2263,6 +2263,51 @@ async fn join_cg(client: &IggyClient) {
.unwrap();
}
+fn assert_balanced_partition_distribution(cg: &ConsumerGroupDetails,
expected_total: u32) {
+ let member_count = cg.members.len() as u32;
+ assert!(member_count > 0, "No members in consumer group");
+
+ let total: u32 = cg.members.iter().map(|m| m.partitions_count).sum();
+ assert_eq!(
+ total, expected_total,
+ "Total partitions mismatch. Expected {expected_total}, got {total}.
Members: {:?}",
+ cg.members
+ );
+
+ let fair_share = expected_total / member_count;
+ let remainder = expected_total % member_count;
+
+ let min_expected = fair_share;
+ let max_expected = if remainder > 0 {
+ fair_share + 1
+ } else {
+ fair_share
+ };
+
+ for member in &cg.members {
+ assert!(
+ member.partitions_count >= min_expected && member.partitions_count
<= max_expected,
+ "Member {} has {} partitions, expected between {} and {}. \
+ Distribution is unbalanced! Members: {:?}",
+ member.id,
+ member.partitions_count,
+ min_expected,
+ max_expected,
+ cg.members
+ );
+ }
+
+ let counts: Vec<u32> = cg.members.iter().map(|m|
m.partitions_count).collect();
Review Comment:
this variance check (lines 2300-2308) is redundant. the per-member range
assert just above (2287-2298) already bounds every member to `[fair_share,
fair_share + (remainder>0?1:0)]`, a span of at most 1, so `max_count -
min_count <= 1` always holds. can drop the whole block plus the `counts` vec.
##########
core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs:
##########
@@ -3267,3 +3312,536 @@ async fn
should_not_complete_other_members_revocations_on_leave(harness: &TestHa
.await
.unwrap();
}
+
+#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
+ heartbeat.enabled = true,
+ heartbeat.interval = "2s",
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true
+))]
+async fn should_distribute_16_partitions_evenly_across_16_consumers(harness:
&TestHarness) {
+ let root_client = harness.root_client().await.unwrap();
+
+ root_client.create_stream(STREAM_NAME).await.unwrap();
Review Comment:
all 6 new tests inline this same ~21-line create stream + topic + consumer
group block, only the partition count differs (16/12/10/16/24/16). also at
3438, 3525, 3610, 3691, 3766. the existing `setup_stream_topic_cg` helper
hardcodes `PARTITIONS_COUNT` so it can't be reused here. a
`setup_stream_topic_cg_with_partitions(client, n)` with the existing helper
delegating to it would cut ~120 lines, and helper-based setup is already the
convention in this file.
--
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]
