fix: re-add consumer group STABLE check in hasLeaderPartition

[waterWang]

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

[vbhanuchander-lang]

Re-adding a readiness gate may well be the right call, but three statements in the stated root cause
do not match the code, and I think they matter because they are the justification for this
particular fix. I posted the same findings on #17193 earlier today
(comment) — apologies if
they crossed with your work.

1. The removed check was on the Connect group, not cg-control. #14395 deleted the
ConsumerGroupState.STABLE guard from hasLeaderPartition, which describes
config.connectGroupId() — the sink's data-topic consumer group — and its purpose was coordinator
leader election (which task owns the first partition). It never inspected the control group, so
restoring it does not "ensure the cg-control consumer has joined".

2. max.poll.interval.ms does not default to 60s — it is 300000 (ConsumerConfig, and the 60s
figure comes from a different setting: WorkerSinkTask.iteration() polls with
Math.max(nextCommit - now, 0), where nextCommit advances by offset.flush.interval.ms, whose
default is 60000L in WorkerConfig. That is a Connect worker setting, not a consumer one.

3. A poll gap does not expire session.timeout.ms. Since KIP-62 the consumer heartbeats from a
background thread, so session.timeout.ms (45000) covers heartbeat loss, not time between poll()
calls; the config that evicts a member for not polling is max.poll.interval.ms, which a ~60s gap
does not breach. Relatedly, task.put(...) is still invoked on every cycle when the topic is idle —
WorkerSinkTask.poll() calls deliverMessages() unconditionally after convertMessages(msgs), with
no non-empty guard — so the control consumer does keep being polled.

None of this says the symptom in #17193 is not real; #11818 reports the same thing. My concern is
narrower: if the mechanism is not the one described, this change may be fixing it by side effect
(delaying commits until the Connect group settles) rather than addressing the cause, and it would be
easy for that to look fixed in one environment and not another.

The thing that would settle it is which group the reporter's UnknownMemberIdException actually
names — cg-control-<uuid> (created per worker at Worker.java:53), <connectGroupId>-coord
(Coordinator.java:97), or the Connect group itself. Those are three different groups with three
different fixes. Worth asking @kuldeep0508 for that line before settling on this approach.

Happy to be shown wrong on any of the three points — they are all checkable against Kafka 3.9
sources.