mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-05-23 10:11:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(kafka): don't let empty leader assignments bypass coverage check

Browse Source 
Review spot: the leader-assignment branch was gated on
`len(request.GroupAssignments) > 0`, so a leader SyncGroup that omitted
every current member (empty array with a non-empty group) fell through
to the server-side-assignment `else` branch and could move the group
Stable without the intended rebalance retry.

Drop the length guard. Whenever the caller is the leader, build the
assigned-member map and run the coverage check; if the assignment
omits any current member (including the all-empty case against a
non-empty group), bump the generation, reset to PreparingRebalance,
clear each member's Assignment, and return REBALANCE_IN_PROGRESS so
the leader rebuilds its snapshot and sends a complete assignment on
retry. The server-side-assignment branch (documented as "should not
happen with Sarama") is now only reachable for non-leader+non-empty
SyncGroups — a genuinely unexpected case — and keeps its existing
warning.
This commit is contained in:
Chris Lu
2026-04-20 15:38:46 -07:00
parent 8ae77f92ec
commit 797f4f779b
1 changed files with 9 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
weed/mq/kafka/protocol/joingroup.go
View File

@@ -924,8 +924,16 @@ func (h *Handler) handleSyncGroup(correlationID uint32, apiVersion uint16, reque
glog.V(2).Infof("[SYNCGROUP] Member=%s Leader=%s GroupState=%s HasAssignments=%v MemberCount=%d Gen=%d", glog.V(2).Infof("[SYNCGROUP] Member=%s Leader=%s GroupState=%s HasAssignments=%v MemberCount=%d Gen=%d",
request.MemberID, group.Leader, group.State, len(request.GroupAssignments) > 0, len(group.Members), request.GenerationID) request.MemberID, group.Leader, group.State, len(request.GroupAssignments) > 0, len(group.Members), request.GenerationID)
if request.MemberID == group.Leader && len(request.GroupAssignments) > 0 { if request.MemberID == group.Leader {
// Leader is providing assignments - process and store them. // Leader is providing assignments - process and store them.
// We don't gate on len(request.GroupAssignments) > 0 here: if the
// leader sends an empty assignments array while the group has
// members, every member is "missing" from the assignment map and
// the coverage check below will reject with REBALANCE_IN_PROGRESS
// — catching the accidental-empty-assignment case instead of
// silently falling through to the server-side-assignment branch
// (which exists only as a "should not happen with Sarama"
// fallback).
// Before committing, verify the leader's assignment covers every // Before committing, verify the leader's assignment covers every
// current member. A late joiner can arrive either during the // current member. A late joiner can arrive either during the