justinmclean opened a new issue, #4391: URL: https://github.com/apache/iggy/issues/4391
### Bug description A member that joins a consumer group while another member's partitions are still pending revocation gets no partitions, and the group never corrects it. Setup: one topic with 4 partitions, one consumer group. Member A owns all 4 and has polled them without committing, so they are in flight. Members B and C then join back to back. Expected: 2, 1, 1. Actual: A=2, B=2, C=0. This stays the same under continuous traffic (60 seconds). `consumer-group get` shows the same thing: ``` | Members count | 3 | | Members | Member id | Partitions count | Partitions | | | 0 | 2 | 0, 1 | | | 1 | 2 | 2, 3 | | | 2 | 0 | | ``` If C joins after B's revocations have completed, the split is correct (2, 1, 1). With nothing in flight, the split is also correct, because the partitions move at join time. This is not #3334. That fix spreads excess over several idle members in one rebalance. Here the second join's rebalance does not see B as needing anything and does not see C as owed anything. ### Cause In `ConsumerGroup::rebalance_cooperative` (core/metadata/src/stm/consumer_group.rs), step 2: - Partitions pending revocation are subtracted from the source's count, so A counts as owning 2. - A member that is the target of a pending revocation is left out of the idle list, so B is skipped. https://github.com/apache/iggy/blob/e4ff95b8852ca0b86efaaa68d62c27d8a2c58fec/core/metadata/src/stm/consumer_group.rs#L229-L254 When C joins, there are 3 members, so fair share is 1 with a remainder of 1. A's effective count of 2 is within `fair_share + 1`, so nothing is collected as excess and C gets nothing. The plan already made for B (two partitions) is not revisited either. The rebalance only runs on join and leave. `complete_revocation` moves the partition to its recorded target and bumps the generation, but does not rebalance again, so the uneven split is permanent until the next join or leave. https://github.com/apache/iggy/blob/e4ff95b8852ca0b86efaaa68d62c27d8a2c58fec/core/metadata/src/stm/consumer_group.rs#L321-L363 ### Affected area / component Iggy server ### Deployment Docker (DockerHub image) ### Versions server 0.9.0 ### Hardware / environment local Mac laptop ### Sample code _No response_ ### Logs _No response_ ### Iggy server config _No response_ ### Reproduction Failing integration test, next to `should_timeout_revocation` in core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs: ```rust #[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server( heartbeat.enabled = true, heartbeat.interval = "60s", consumer_group.rebalancing_timeout = "3s" ))] async fn should_balance_member_that_joins_while_revocation_pending(harness: &TestHarness) { const PARTITIONS: u32 = 4; let root_client = harness.root_client().await.unwrap(); setup_stream_topic_cg_with_partitions(&root_client, PARTITIONS).await; for partition_id in 0..PARTITIONS { let message = IggyMessage::from_str(&format!("msg-p{partition_id}")).unwrap(); let mut messages = vec![message]; root_client .send_messages( &Identifier::named(STREAM_NAME).unwrap(), &Identifier::named(TOPIC_NAME).unwrap(), &Partitioning::partition_id(partition_id), &mut messages, ) .await .unwrap(); } // 1. Consumer1 joins, gets all 4 partitions and polls each without // committing, so every partition is in flight. let client1 = harness.new_client().await.unwrap(); client1 .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD) .await .unwrap(); join_cg(&client1).await; let consumer = Consumer::group(Identifier::named(CONSUMER_GROUP_NAME).unwrap()); for _ in 0..PARTITIONS { client1 .poll_messages( &Identifier::named(STREAM_NAME).unwrap(), &Identifier::named(TOPIC_NAME).unwrap(), None, &consumer, &PollingStrategy::next(), 1, false, ) .await .unwrap(); } // 2. Consumer2 joins: 2 partitions are pending revocation toward it. let client2 = harness.new_client().await.unwrap(); client2 .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD) .await .unwrap(); join_cg(&client2).await; // 3. Consumer3 joins before those revocations complete. let client3 = harness.new_client().await.unwrap(); client3 .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD) .await .unwrap(); join_cg(&client3).await; // 4. Wait past the revocation timeout (3s) + check interval (1s) + margin. sleep(Duration::from_secs(6)).await; // 5. With 4 partitions and 3 members the split must be 2/1/1. let cg = get_consumer_group(&root_client).await; assert_eq!(cg.members_count, 3); assert_unique_partition_assignments(&cg); assert_balanced_partition_distribution(&cg, PARTITIONS); // Cleanup root_client .delete_stream(&Identifier::named(STREAM_NAME).unwrap()) .await .unwrap(); } ``` It fails on all three transports, every run (9 of 9): ``` Member 2 has 0 partitions, expected between 1 and 2. Distribution is unbalanced! Members: [{ id: 0, partitions: [0, 1] }, { id: 1, partitions: [3, 2] }, { id: 2, partitions: [] }] ``` The same steps with the Java SDK 0.9.0 against the apache/iggy:0.9.0 image give the same result. In that run A commits its batch after C joins, so the revocations complete on commit, not on timeout. The image needs `IGGY_NODE_ADVERTISED_ADDRESS=127.0.0.1` when `IGGY_TCP_ADDRESS` binds `0.0.0.0`. Possible fix: Rebalance again when a revocation completes. After `complete_revocation` moves the partition, run the cooperative pass again. The apply has no in-flight data, so it would treat every owned partition as in flight and pending-revoke the excess. The reconciler already completes a never-polled or drained partition on its next pass, so the cost is one extra tick. This also covers any other case where a pending plan goes stale. ### Contribution - [ ] I'm willing to submit a pull request to fix this bug ### Good first issue - [ ] I think this could be a good first issue for a new contributor -- 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]
