fix(kafka): evict expired group members on rejoin to unblock fast restart (#9243)
fix(kafka): evict expired group members on JoinGroup and during cleanup Phantom members lingering past their session timeout were blocking TestOffsetManagement/ConsumerGroupResumption: the cleanup goroutine runs every 30s but session timeouts can be as short as 6s, so a lost LeaveGroup or ungraceful disconnect could leave a dead member holding the leader slot or a partition until the next cleanup tick. New consumers rejoining within that window saw stale group composition and burned all 5 test-level join retries (~118s, generation churned to 22 in the failing CI run). Add EvictExpiredMembersLocked on GroupCoordinator and call it from handleJoinGroup so each rejoin sees a fresh member list. Refactor performCleanup to share the helper, and on either path move surviving members to PreparingRebalance, bump generation, clear cached assignments (so the non-leader SyncGroup path returns REBALANCE_IN_PROGRESS instead of serving stale partitions), re-elect a leader from survivors when needed, and rebuild SubscribedTopics so topics only the evicted member subscribed to are dropped.
This commit is contained in:
Chris Lu
GitHub
parent
fa492a9eed
commit
3f5b4814b7
@ -297,6 +297,36 @@ func (gc *GroupCoordinator) cleanupRoutine() {
|
||||
}
|
||||
}
|
||||
|
||||
// EvictExpiredMembersLocked removes members whose session has expired and
|
||||
// returns their IDs. The caller is responsible for any group-state transition
|
||||
// that follows (e.g. moving the group to GroupStateEmpty when no members
|
||||
// remain, or kicking off a rebalance among the survivors). The caller must
|
||||
// hold group.Mu.
|
||||
func (gc *GroupCoordinator) EvictExpiredMembersLocked(group *ConsumerGroup) []string {
|
||||
now := time.Now()
|
||||
var expired []string
|
||||
for memberID, member := range group.Members {
|
||||
// Defensive: a SessionTimeout of zero means the JoinGroup payload
|
||||
// hasn't populated it yet, not "expired immediately".
|
||||
if member.SessionTimeout <= 0 {
|
||||
continue
|
||||
}
|
||||
sessionDuration := time.Duration(member.SessionTimeout) * time.Millisecond
|
||||
if now.Sub(member.LastHeartbeat) <= sessionDuration {
|
||||
continue
|
||||
}
|
||||
if member.GroupInstanceID != nil && *member.GroupInstanceID != "" {
|
||||
delete(group.StaticMembers, *member.GroupInstanceID)
|
||||
}
|
||||
delete(group.Members, memberID)
|
||||
if group.Leader == memberID {
|
||||
group.Leader = ""
|
||||
}
|
||||
expired = append(expired, memberID)
|
||||
}
|
||||
return expired
|
||||
}
|
||||
|
||||
// performCleanup removes expired members and empty groups
|
||||
func (gc *GroupCoordinator) performCleanup() {
|
||||
now := time.Now()
|
||||
@ -310,21 +340,38 @@ func (gc *GroupCoordinator) performCleanup() {
|
||||
for groupID, group := range gc.groups {
|
||||
group.Mu.Lock()
|
||||
|
||||
// Check for expired members (session timeout)
|
||||
expiredMembers := make([]string, 0)
|
||||
for memberID, member := range group.Members {
|
||||
sessionDuration := time.Duration(member.SessionTimeout) * time.Millisecond
|
||||
timeSinceHeartbeat := now.Sub(member.LastHeartbeat)
|
||||
if timeSinceHeartbeat > sessionDuration {
|
||||
expiredMembers = append(expiredMembers, memberID)
|
||||
}
|
||||
}
|
||||
// Evict expired members (session timeout). EvictExpiredMembersLocked
|
||||
// also handles static-member unregistration and clears group.Leader if
|
||||
// the leader was evicted.
|
||||
expired := gc.EvictExpiredMembersLocked(group)
|
||||
|
||||
// Remove expired members
|
||||
for _, memberID := range expiredMembers {
|
||||
delete(group.Members, memberID)
|
||||
if group.Leader == memberID {
|
||||
group.Leader = ""
|
||||
// If there are surviving members, mirror the join-time eviction path:
|
||||
// move the group to PreparingRebalance, bump the generation, clear
|
||||
// cached assignments, and select a new leader if one was evicted —
|
||||
// otherwise the partitions owned by the expired member would stay
|
||||
// assigned to that ghost slot until some unrelated join/leave bumped
|
||||
// the group out of Stable. The join-time path doesn't need to pick a
|
||||
// leader because it has an incoming member; here we have to.
|
||||
if len(expired) > 0 && len(group.Members) > 0 {
|
||||
if group.Leader == "" {
|
||||
for memberID := range group.Members {
|
||||
group.Leader = memberID
|
||||
break
|
||||
}
|
||||
}
|
||||
group.State = GroupStatePreparingRebalance
|
||||
group.Generation++
|
||||
for _, m := range group.Members {
|
||||
m.State = MemberStatePending
|
||||
m.Assignment = nil
|
||||
}
|
||||
// Rebuild subscribed topics from remaining members so a topic
|
||||
// only the evicted member subscribed to is dropped.
|
||||
group.SubscribedTopics = make(map[string]bool)
|
||||
for _, m := range group.Members {
|
||||
for _, topic := range m.Subscription {
|
||||
group.SubscribedTopics[topic] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -228,3 +228,170 @@ func TestGroupCoordinator_GenerateMemberID(t *testing.T) {
|
||||
t.Errorf("Expected member ID to start with 'consumer-', got: %s", id1)
|
||||
}
|
||||
}
|
||||
|
||||
// TestGroupCoordinator_EvictExpiredMembersLocked covers the on-rejoin eviction
|
||||
// path: JoinGroup calls this helper with the group lock held to drop phantom
|
||||
// members whose session has expired but whose LeaveGroup never landed (the
|
||||
// 30s cleanup tick is too coarse for fast consumer restarts with a 6s session
|
||||
// timeout — see TestOffsetManagement/ConsumerGroupResumption).
|
||||
func TestGroupCoordinator_EvictExpiredMembersLocked(t *testing.T) {
|
||||
gc := NewGroupCoordinator()
|
||||
defer gc.Close()
|
||||
|
||||
group := gc.GetOrCreateGroup("test-group")
|
||||
|
||||
expiredLeader := &GroupMember{
|
||||
ID: "expired-leader",
|
||||
SessionTimeout: 6000, // 6s
|
||||
LastHeartbeat: time.Now().Add(-10 * time.Second), // 10s ago: expired
|
||||
State: MemberStateStable,
|
||||
}
|
||||
staticInstance := "static-1"
|
||||
expiredStatic := &GroupMember{
|
||||
ID: "expired-static",
|
||||
SessionTimeout: 6000,
|
||||
LastHeartbeat: time.Now().Add(-30 * time.Second),
|
||||
GroupInstanceID: &staticInstance,
|
||||
State: MemberStateStable,
|
||||
}
|
||||
healthyMember := &GroupMember{
|
||||
ID: "healthy-member",
|
||||
SessionTimeout: 30000,
|
||||
LastHeartbeat: time.Now(),
|
||||
State: MemberStateStable,
|
||||
}
|
||||
|
||||
group.Mu.Lock()
|
||||
group.Members[expiredLeader.ID] = expiredLeader
|
||||
group.Members[expiredStatic.ID] = expiredStatic
|
||||
group.Members[healthyMember.ID] = healthyMember
|
||||
group.StaticMembers[staticInstance] = expiredStatic.ID
|
||||
group.Leader = expiredLeader.ID
|
||||
|
||||
evicted := gc.EvictExpiredMembersLocked(group)
|
||||
defer group.Mu.Unlock()
|
||||
|
||||
if len(evicted) != 2 {
|
||||
t.Fatalf("expected 2 evictions, got %d (%v)", len(evicted), evicted)
|
||||
}
|
||||
if _, exists := group.Members[expiredLeader.ID]; exists {
|
||||
t.Errorf("expired leader should have been removed")
|
||||
}
|
||||
if _, exists := group.Members[expiredStatic.ID]; exists {
|
||||
t.Errorf("expired static member should have been removed")
|
||||
}
|
||||
if _, exists := group.Members[healthyMember.ID]; !exists {
|
||||
t.Errorf("healthy member should remain")
|
||||
}
|
||||
if group.Leader != "" {
|
||||
t.Errorf("expected leader cleared after evicting old leader, got %q", group.Leader)
|
||||
}
|
||||
if _, exists := group.StaticMembers[staticInstance]; exists {
|
||||
t.Errorf("expired static instance ID should have been unregistered")
|
||||
}
|
||||
}
|
||||
|
||||
// TestGroupCoordinator_Cleanup_SurvivorsRebalance covers the survivors path:
|
||||
// when expired members are evicted but at least one healthy member remains,
|
||||
// performCleanup must move the group to PreparingRebalance, bump the
|
||||
// generation, clear cached assignments, and elect a new leader. Otherwise
|
||||
// partitions held by the evicted member would stay assigned to the ghost
|
||||
// slot until some unrelated join/leave bumped the group out of Stable.
|
||||
func TestGroupCoordinator_Cleanup_SurvivorsRebalance(t *testing.T) {
|
||||
gc := NewGroupCoordinator()
|
||||
defer gc.Close()
|
||||
|
||||
group := gc.GetOrCreateGroup("test-group")
|
||||
|
||||
expiredLeader := &GroupMember{
|
||||
ID: "expired-leader",
|
||||
SessionTimeout: 1000, // 1s
|
||||
LastHeartbeat: time.Now().Add(-5 * time.Second), // expired
|
||||
Subscription: []string{"shared-topic", "leader-only"},
|
||||
Assignment: []PartitionAssignment{{Topic: "shared-topic", Partition: 0}},
|
||||
State: MemberStateStable,
|
||||
}
|
||||
survivor := &GroupMember{
|
||||
ID: "survivor",
|
||||
SessionTimeout: 30000,
|
||||
LastHeartbeat: time.Now(),
|
||||
Subscription: []string{"shared-topic"},
|
||||
Assignment: []PartitionAssignment{{Topic: "shared-topic", Partition: 1}},
|
||||
State: MemberStateStable,
|
||||
}
|
||||
|
||||
group.Mu.Lock()
|
||||
group.Members[expiredLeader.ID] = expiredLeader
|
||||
group.Members[survivor.ID] = survivor
|
||||
group.Leader = expiredLeader.ID
|
||||
group.State = GroupStateStable
|
||||
group.Generation = 5
|
||||
group.SubscribedTopics = map[string]bool{
|
||||
"shared-topic": true,
|
||||
"leader-only": true,
|
||||
}
|
||||
group.Mu.Unlock()
|
||||
|
||||
gc.performCleanup()
|
||||
|
||||
group.Mu.RLock()
|
||||
defer group.Mu.RUnlock()
|
||||
|
||||
if _, exists := group.Members[expiredLeader.ID]; exists {
|
||||
t.Errorf("expired leader should have been evicted")
|
||||
}
|
||||
if _, exists := group.Members[survivor.ID]; !exists {
|
||||
t.Fatalf("survivor should remain in group")
|
||||
}
|
||||
if group.Leader != survivor.ID {
|
||||
t.Errorf("expected leader re-elected to %q, got %q", survivor.ID, group.Leader)
|
||||
}
|
||||
if group.State != GroupStatePreparingRebalance {
|
||||
t.Errorf("expected state PreparingRebalance after eviction with survivors, got %s", group.State)
|
||||
}
|
||||
if group.Generation != 6 {
|
||||
t.Errorf("expected generation 6, got %d", group.Generation)
|
||||
}
|
||||
if survivor.State != MemberStatePending {
|
||||
t.Errorf("expected survivor state Pending, got %s", survivor.State)
|
||||
}
|
||||
if len(survivor.Assignment) != 0 {
|
||||
t.Errorf("expected survivor assignment cleared so non-leader SyncGroup returns REBALANCE_IN_PROGRESS, got %v", survivor.Assignment)
|
||||
}
|
||||
if _, ok := group.SubscribedTopics["leader-only"]; ok {
|
||||
t.Errorf("expected topic only the evicted member subscribed to be dropped from SubscribedTopics")
|
||||
}
|
||||
if _, ok := group.SubscribedTopics["shared-topic"]; !ok {
|
||||
t.Errorf("expected shared-topic to remain in SubscribedTopics")
|
||||
}
|
||||
}
|
||||
|
||||
// TestGroupCoordinator_EvictExpiredMembersLocked_ZeroSessionTimeout makes sure
|
||||
// a brand-new member that hasn't yet had its SessionTimeout populated isn't
|
||||
// auto-evicted (zero is not "expired immediately").
|
||||
func TestGroupCoordinator_EvictExpiredMembersLocked_ZeroSessionTimeout(t *testing.T) {
|
||||
gc := NewGroupCoordinator()
|
||||
defer gc.Close()
|
||||
|
||||
group := gc.GetOrCreateGroup("test-group")
|
||||
|
||||
pristine := &GroupMember{
|
||||
ID: "pristine",
|
||||
SessionTimeout: 0, // not yet populated
|
||||
LastHeartbeat: time.Time{},
|
||||
State: MemberStatePending,
|
||||
}
|
||||
|
||||
group.Mu.Lock()
|
||||
group.Members[pristine.ID] = pristine
|
||||
|
||||
evicted := gc.EvictExpiredMembersLocked(group)
|
||||
defer group.Mu.Unlock()
|
||||
|
||||
if len(evicted) != 0 {
|
||||
t.Errorf("member with zero SessionTimeout should not be evicted, got %v", evicted)
|
||||
}
|
||||
if _, exists := group.Members[pristine.ID]; !exists {
|
||||
t.Errorf("pristine member should remain in group")
|
||||
}
|
||||
}
|
||||
|
||||
@ -78,6 +78,41 @@ func (h *Handler) handleJoinGroup(connContext *ConnectionContext, correlationID
|
||||
// Update group's last activity
|
||||
group.LastActivity = time.Now()
|
||||
|
||||
// Evict members whose session has expired before processing this join.
|
||||
// The cleanup goroutine runs every 30s, but session timeouts can be as
|
||||
// short as 6s — so a lost LeaveGroup or an ungraceful disconnect can leave
|
||||
// a phantom member in the group long enough for the next rejoin to land
|
||||
// in a stale composition. Sweeping here forces the new join into a clean
|
||||
// view of the group and unblocks fast restart-and-rejoin patterns
|
||||
// (TestOffsetManagement/ConsumerGroupResumption was burning all 5
|
||||
// test-level retries because the dead previous member kept it from
|
||||
// becoming leader / making progress).
|
||||
if expired := h.groupCoordinator.EvictExpiredMembersLocked(group); len(expired) > 0 {
|
||||
glog.V(1).Infof("[JoinGroup] evicted %d expired member(s) from group %s on rejoin: %v",
|
||||
len(expired), request.GroupID, expired)
|
||||
h.updateGroupSubscription(group)
|
||||
if len(group.Members) == 0 {
|
||||
// Mark the group Empty without bumping generation — the
|
||||
// Empty/Stable case in the state-machine switch below will do the
|
||||
// bump as part of this new member joining.
|
||||
group.State = consumer.GroupStateEmpty
|
||||
group.Leader = ""
|
||||
} else {
|
||||
// Surviving members must rebalance to drop the dead member's
|
||||
// partitions. Bump here so the PreparingRebalance branch in the
|
||||
// switch below is a no-op (avoids a double generation bump) and
|
||||
// clear cached assignments so the non-leader SyncGroup path
|
||||
// returns REBALANCE_IN_PROGRESS instead of serving stale
|
||||
// pre-eviction partitions to a rejoiner.
|
||||
group.State = consumer.GroupStatePreparingRebalance
|
||||
group.Generation++
|
||||
for _, m := range group.Members {
|
||||
m.State = consumer.MemberStatePending
|
||||
m.Assignment = nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Handle member ID logic with static membership support
|
||||
var memberID string
|
||||
var isNewMember bool
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
