diff --git a/weed/mq/kafka/consumer/group_coordinator.go b/weed/mq/kafka/consumer/group_coordinator.go index 3bccc5da0..6be0693e4 100644 --- a/weed/mq/kafka/consumer/group_coordinator.go +++ b/weed/mq/kafka/consumer/group_coordinator.go @@ -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 + } } } diff --git a/weed/mq/kafka/consumer/group_coordinator_test.go b/weed/mq/kafka/consumer/group_coordinator_test.go index 5be4f7f93..b59a840b7 100644 --- a/weed/mq/kafka/consumer/group_coordinator_test.go +++ b/weed/mq/kafka/consumer/group_coordinator_test.go @@ -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") + } +} diff --git a/weed/mq/kafka/protocol/joingroup.go b/weed/mq/kafka/protocol/joingroup.go index 918d57470..32ec21afe 100644 --- a/weed/mq/kafka/protocol/joingroup.go +++ b/weed/mq/kafka/protocol/joingroup.go @@ -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