waterWang opened a new pull request, #17639:
URL: https://github.com/apache/iceberg/pull/17639
## Fix
Re-add the `ConsumerGroupState.STABLE` check in `hasLeaderPartition()` that
was removed in PR #14395.
### Root cause
PR #14395 ("Don't check that consumer group is stable for coordinator leader
election") removed the `groupDesc.state() == ConsumerGroupState.STABLE` guard
from `hasLeaderPartition()`. Without this check, the coordinator can start
firing commits before the `cg-control` consumer group has stabilized. When the
data topic goes idle after initial consumption, the main consumer's `poll()`
blocks for `max.poll.interval.ms` (default 60s), during which the `cg-control`
consumer cannot heartbeat, causing its session timeout (45s) to expire and the
group to never reach STABLE.
### Fix
Restore the STABLE state check: only consider this worker the leader when
the consumer group is in the `STABLE` state, ensuring the `cg-control` consumer
has joined before the coordinator starts commit cycles.
### Changes
1. **CommitterImpl.java**: Added `import
org.apache.kafka.common.ConsumerGroupState` and re-added `if (groupDesc.state()
== ConsumerGroupState.STABLE)` guard
2. **TestCommitterImpl.java**: Added `ConsumerGroupState.STABLE` mock to
`testHasLeaderPartition`, added `testHasLeaderPartitionNotStable` test for the
non-STABLE case
Fixes #17193
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
