feat(#3266): reset consumer group offsets while keeping consumers 'online'

[superhx]

close #3266

[daniel-y]

Code review

Found 1 issue:

1. forceRemoveAllClassicGroupMembers makes an invalid state transition: group.transitionTo(ClassicGroupState.EMPTY) is called directly, but ClassicGroupState.EMPTY only accepts PREPARING_REBALANCE as a valid previous state. When the group is in STABLE or COMPLETING_REBALANCE (the normal state for a group with active consumers — which is the primary use case for this feature), assertValidTransition will throw IllegalStateException and crash the coordinator shard. The normal leave flow always transitions through PREPARING_REBALANCE before reaching EMPTY.

automq/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

Lines 5780 to 5786 in a9ae1c6

	for (String memberId : new ArrayList<>(group.allMemberIds())) {
	timer.cancel(classicGroupHeartbeatKey(group.groupId(), memberId));
	group.remove(memberId);
	}
	group.transitionTo(ClassicGroupState.EMPTY);
	records.add(CoordinatorRecordHelpers.newEmptyGroupMetadataRecord(group, metadataImage.features().metadataVersion()));
	}

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}

[superhx]

Code review

Found 1 issue:

1. forceRemoveAllClassicGroupMembers makes an invalid state transition: group.transitionTo(ClassicGroupState.EMPTY) is called directly, but ClassicGroupState.EMPTY only accepts PREPARING_REBALANCE as a valid previous state. When the group is in STABLE or COMPLETING_REBALANCE (the normal state for a group with active consumers — which is the primary use case for this feature), assertValidTransition will throw IllegalStateException and crash the coordinator shard. The normal leave flow always transitions through PREPARING_REBALANCE before reaching EMPTY.

automq/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

Lines 5780 to 5786 in a9ae1c6

	for (String memberId : new ArrayList<>(group.allMemberIds())) {
	timer.cancel(classicGroupHeartbeatKey(group.groupId(), memberId));
	group.remove(memberId);
	}
	group.transitionTo(ClassicGroupState.EMPTY);
	records.add(CoordinatorRecordHelpers.newEmptyGroupMetadataRecord(group, metadataImage.features().metadataVersion()));
	}

🤖 Generated with Claude Code

• If this code review was useful, please react with 👍. Otherwise, react with 👎.

Fixed. This issue only happens when the new coordinator is enabled.

[daniel-y]

The fix at e288feb handles STABLE and COMPLETING_REBALANCE, but PREPARING_REBALANCE -> PREPARING_REBALANCE is still an invalid transition per the state machine:

PREPARING_REBALANCE.addValidPreviousStates(STABLE, COMPLETING_REBALANCE, EMPTY);

If a rebalance is already in progress when the force commit arrives, group.transitionTo(ClassicGroupState.PREPARING_REBALANCE) will throw IllegalStateException.

automq/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

Lines 5784 to 5787 in e288feb

	// EMPTY only accepts PREPARING_REBALANCE as a valid previous state,
	// so we must transition through it first (mirroring the normal leave flow).
	group.transitionTo(ClassicGroupState.PREPARING_REBALANCE);
	group.transitionTo(ClassicGroupState.EMPTY);

Suggested fix — guard the transition:

if (!group.isInState(ClassicGroupState.PREPARING_REBALANCE)) {
    group.transitionTo(ClassicGroupState.PREPARING_REBALANCE);
}
group.transitionTo(ClassicGroupState.EMPTY);

🤖 Generated with Claude Code

[superhx]

The fix at e288feb handles STABLE and COMPLETING_REBALANCE, but PREPARING_REBALANCE -> PREPARING_REBALANCE is still an invalid transition per the state machine:

PREPARING_REBALANCE.addValidPreviousStates(STABLE, COMPLETING_REBALANCE, EMPTY);

If a rebalance is already in progress when the force commit arrives, group.transitionTo(ClassicGroupState.PREPARING_REBALANCE) will throw IllegalStateException.

automq/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

Lines 5784 to 5787 in e288feb

	// EMPTY only accepts PREPARING_REBALANCE as a valid previous state,
	// so we must transition through it first (mirroring the normal leave flow).
	group.transitionTo(ClassicGroupState.PREPARING_REBALANCE);
	group.transitionTo(ClassicGroupState.EMPTY);

Suggested fix — guard the transition:

if (!group.isInState(ClassicGroupState.PREPARING_REBALANCE)) {
    group.transitionTo(ClassicGroupState.PREPARING_REBALANCE);
}
group.transitionTo(ClassicGroupState.EMPTY);

🤖 Generated with Claude Code

Fixed