From 86c5e815d202f0dca07aea29a90712a23fb56cd5 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 20 Apr 2026 10:11:45 -0700 Subject: [PATCH] fix(kafka): make consumer-group rebalancing work end-to-end (#9143) * fix(kafka): make consumer-group rebalancing work end-to-end TestConsumerGroups was failing every run since the job was added (2026-04-17) but the failures were masked by a `|| echo ...` trailer on the go test invocation, so the CI reported green. Removing the mask exposes several real bugs in the gateway's group-coordinator code: 1. JoinGroup deduplicated members by ClientID, which collapsed two Sarama consumers that share the default ClientID ("sarama") into a single member slot and broke rebalancing. Key dedup off the TCP ConnectionID instead; keep ClientID on the member for DescribeGroup fidelity. 2. Every JoinGroup replaced the *GroupMember struct, wiping the Assignment the leader had just published in its SyncGroup and leaving non-leader consumers with 0 partitions after a rebalance. Update the existing member in place on rejoin. 3. Non-leader SyncGroup returned an empty assignment while the leader was mid-rebalance, so consumers silently came up with no partitions. Return REBALANCE_IN_PROGRESS when the group is not Stable so Sarama retries the join/sync cycle (4 retries x 2s backoff by default). 4. Heartbeat returned ILLEGAL_GENERATION on a gen mismatch even when the group was in PreparingRebalance/CompletingRebalance. Return REBALANCE_IN_PROGRESS in that case so the heartbeat loop cleanly cancels the session instead of tearing it down on a fatal error. 5. LeaveGroup parser only handled v0-v2. Sarama at V2_8_0_0 sends v3 (Members array) by default, so the gateway silently rejected the request as InvalidGroupID and dead consumers stayed in the group as phantom leaders. Added v3 (Members array) and v4+ (flexible/compact/ tagged-fields) parsing. The rebalancing integration tests called Consume() once per consumer, which cannot survive a rebalance (heartbeat RBIP cancels the session and Consume() returns - this is documented Sarama behaviour; callers are expected to loop). Added a runConsumeLoop helper and used it in the four affected sub-tests. RebalanceTestHandler.Setup now overwrites stale entries in its assignments channel so the test observes the settled post-rebalance snapshot rather than whatever arrived first. * fix(kafka): address PR review feedback - JoinGroup now snapshots existing members before mutating and restores the snapshot on INCONSISTENT_GROUP_PROTOCOL rollback. Previously the rollback path always deleted the entry, corrupting group state when an existing member rejoined with an incompatible protocol. - handleLeaveGroup iterates request.Members instead of processing only the first entry, so v3+ batch departures (KIP-345 style) correctly remove every listed member and build a per-member response. A single group-state transition runs after the loop, with leader election only triggered if the actual group leader was among the departures. - Added buildLeaveGroupFlexibleResponse for v4+ clients. The parser already decoded flexible versions, but the response still went out in non-flexible encoding (4-byte array lengths, 2-byte strings, no tagged fields), which v4+ clients could not parse. Route flexible versions through the new builder; v1-v3 keep buildLeaveGroupFullResponse. - BasicFunctionality gives each consumer its own ConsumerGroupHandler/ready channel. The previous shared handler closed ready once, so readyCount advanced to numConsumers from a single signal; the test could proceed without the other consumers actually reaching Setup. - RebalanceTestHandler.assignments is now a size-1 channel, so readers always observe the most recent rebalance snapshot instead of an intermediate one from an earlier round. --- .github/workflows/kafka-tests.yml | 2 +- .../kafka/integration/consumer_groups_test.go | 45 ++- test/kafka/integration/rebalancing_test.go | 115 +++--- weed/mq/kafka/consumer/group_coordinator.go | 3 +- .../kafka/protocol/consumer_coordination.go | 379 ++++++++++++++---- weed/mq/kafka/protocol/joingroup.go | 108 +++-- 6 files changed, 457 insertions(+), 195 deletions(-) diff --git a/.github/workflows/kafka-tests.yml b/.github/workflows/kafka-tests.yml index 684ff0713..81205d6ef 100644 --- a/.github/workflows/kafka-tests.yml +++ b/.github/workflows/kafka-tests.yml @@ -445,7 +445,7 @@ jobs: # Test consumer group functionality with explicit timeout ulimit -n 512 || echo "Warning: Could not set file descriptor limit" ulimit -u 100 || echo "Warning: Could not set process limit" - timeout 240s go test -v -run "^TestConsumerGroups" -timeout 180s ./integration/... || echo "Test execution timed out or failed" + timeout 240s go test -v -run "^TestConsumerGroups" -timeout 180s ./integration/... env: GOMAXPROCS: 1 SEAWEEDFS_MASTERS: 127.0.0.1:9333 diff --git a/test/kafka/integration/consumer_groups_test.go b/test/kafka/integration/consumer_groups_test.go index 5407a2999..4456c77e0 100644 --- a/test/kafka/integration/consumer_groups_test.go +++ b/test/kafka/integration/consumer_groups_test.go @@ -49,12 +49,20 @@ func testConsumerGroupBasicFunctionality(t *testing.T, addr string) { err = client.ProduceMessages(topicName, messages) testutil.AssertNoError(t, err, "Failed to produce messages") - // Test with multiple consumers in the same group + // Test with multiple consumers in the same group. The messages channel is + // shared so the assertion can verify total-consumed/no-duplicates across the + // group. Each consumer gets its own handler (and its own ready channel) so + // we can wait for each distinct consumer to reach Setup, rather than + // relying on a single close() that only signals once. numConsumers := 3 - handler := &ConsumerGroupHandler{ - messages: make(chan *sarama.ConsumerMessage, len(messages)), - ready: make(chan bool), - t: t, + handlers := make([]*ConsumerGroupHandler, numConsumers) + sharedMessages := make(chan *sarama.ConsumerMessage, len(messages)) + for i := 0; i < numConsumers; i++ { + handlers[i] = &ConsumerGroupHandler{ + messages: sharedMessages, + ready: make(chan bool), + t: t, + } } var wg sync.WaitGroup @@ -72,35 +80,32 @@ func testConsumerGroupBasicFunctionality(t *testing.T, addr string) { } defer consumerGroup.Close() - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) defer cancel() - err = consumerGroup.Consume(ctx, []string{topicName}, handler) - if err != nil && err != context.DeadlineExceeded { - consumerErrors <- fmt.Errorf("consumer %d: consumption error: %v", consumerID, err) - return - } + runConsumeLoop(t, ctx, fmt.Sprintf("Consumer%d", consumerID), + consumerGroup, []string{topicName}, handlers[consumerID]) }(i) } - // Wait for consumers to be ready - readyCount := 0 - for readyCount < numConsumers { + // Wait for each consumer to be ready. Multi-consumer rebalance can take + // a few heartbeat intervals (default 3s) as the initial leader receives + // REBALANCE_IN_PROGRESS from its heartbeat and re-joins, so allow headroom. + for i := 0; i < numConsumers; i++ { select { - case <-handler.ready: - readyCount++ - case <-time.After(5 * time.Second): - t.Fatalf("Timeout waiting for consumers to be ready") + case <-handlers[i].ready: + case <-time.After(20 * time.Second): + t.Fatalf("Timeout waiting for consumer %d to be ready", i) } } // Collect consumed messages consumedMessages := make([]*sarama.ConsumerMessage, 0, len(messages)) - messageTimeout := time.After(10 * time.Second) + messageTimeout := time.After(15 * time.Second) for len(consumedMessages) < len(messages) { select { - case msg := <-handler.messages: + case msg := <-sharedMessages: consumedMessages = append(consumedMessages, msg) case err := <-consumerErrors: t.Fatalf("Consumer error: %v", err) diff --git a/test/kafka/integration/rebalancing_test.go b/test/kafka/integration/rebalancing_test.go index fb962f888..23cd08f8f 100644 --- a/test/kafka/integration/rebalancing_test.go +++ b/test/kafka/integration/rebalancing_test.go @@ -2,6 +2,7 @@ package integration import ( "context" + "errors" "fmt" "sync" "testing" @@ -11,6 +12,25 @@ import ( "github.com/seaweedfs/seaweedfs/test/kafka/internal/testutil" ) +// runConsumeLoop drives consumerGroup.Consume in a for-loop until ctx is cancelled +// or the group is closed. This is the idiomatic Sarama pattern: Consume returns +// when a rebalance cancels the session, and the caller is expected to re-enter. +// Tests that observe rebalance behaviour MUST loop — otherwise the consumer goroutine +// exits on the first rebalance and never sees the post-rebalance assignment. +func runConsumeLoop(t *testing.T, ctx context.Context, name string, cg sarama.ConsumerGroup, topics []string, handler sarama.ConsumerGroupHandler) { + for { + if err := cg.Consume(ctx, topics, handler); err != nil { + if ctx.Err() != nil || errors.Is(err, sarama.ErrClosedConsumerGroup) { + return + } + t.Logf("%s Consume returned: %v (will retry)", name, err) + } + if ctx.Err() != nil { + return + } + } +} + func testSingleConsumerAllPartitions(t *testing.T, addr, topicName, groupID string) { config := sarama.NewConfig() config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange @@ -28,7 +48,7 @@ func testSingleConsumerAllPartitions(t *testing.T, addr, topicName, groupID stri handler := &RebalanceTestHandler{ messages: make(chan *sarama.ConsumerMessage, 20), ready: make(chan bool), - assignments: make(chan []int32, 5), + assignments: make(chan []int32, 1), t: t, } @@ -93,7 +113,7 @@ func testTwoConsumersRebalance(t *testing.T, addr, topicName, groupID string) { handler1 := &RebalanceTestHandler{ messages: make(chan *sarama.ConsumerMessage, 20), ready: make(chan bool), - assignments: make(chan []int32, 5), + assignments: make(chan []int32, 1), t: t, name: "Consumer1", } @@ -101,12 +121,7 @@ func testTwoConsumersRebalance(t *testing.T, addr, topicName, groupID string) { ctx1, cancel1 := context.WithTimeout(context.Background(), 45*time.Second) defer cancel1() - go func() { - err := consumerGroup1.Consume(ctx1, []string{topicName}, handler1) - if err != nil && err != context.DeadlineExceeded { - t.Logf("Consumer1 error: %v", err) - } - }() + go runConsumeLoop(t, ctx1, "Consumer1", consumerGroup1, []string{topicName}, handler1) // Wait for first consumer to be ready and get initial assignment <-handler1.ready @@ -132,7 +147,7 @@ func testTwoConsumersRebalance(t *testing.T, addr, topicName, groupID string) { handler2 := &RebalanceTestHandler{ messages: make(chan *sarama.ConsumerMessage, 20), ready: make(chan bool), - assignments: make(chan []int32, 5), + assignments: make(chan []int32, 1), t: t, name: "Consumer2", } @@ -140,12 +155,7 @@ func testTwoConsumersRebalance(t *testing.T, addr, topicName, groupID string) { ctx2, cancel2 := context.WithTimeout(context.Background(), 30*time.Second) defer cancel2() - go func() { - err := consumerGroup2.Consume(ctx2, []string{topicName}, handler2) - if err != nil && err != context.DeadlineExceeded { - t.Logf("Consumer2 error: %v", err) - } - }() + go runConsumeLoop(t, ctx2, "Consumer2", consumerGroup2, []string{topicName}, handler2) // Wait for second consumer to be ready <-handler2.ready @@ -228,7 +238,7 @@ func testConsumerLeaveRebalance(t *testing.T, addr, topicName, groupID string) { handler1 := &RebalanceTestHandler{ messages: make(chan *sarama.ConsumerMessage, 20), ready: make(chan bool), - assignments: make(chan []int32, 5), + assignments: make(chan []int32, 1), t: t, name: "Consumer1", } @@ -236,7 +246,7 @@ func testConsumerLeaveRebalance(t *testing.T, addr, topicName, groupID string) { handler2 := &RebalanceTestHandler{ messages: make(chan *sarama.ConsumerMessage, 20), ready: make(chan bool), - assignments: make(chan []int32, 5), + assignments: make(chan []int32, 1), t: t, name: "Consumer2", } @@ -246,20 +256,9 @@ func testConsumerLeaveRebalance(t *testing.T, addr, topicName, groupID string) { ctx2, cancel2 := context.WithTimeout(context.Background(), 30*time.Second) - // Start both consumers - go func() { - err := consumerGroup1.Consume(ctx1, []string{topicName}, handler1) - if err != nil && err != context.DeadlineExceeded { - t.Logf("Consumer1 error: %v", err) - } - }() - - go func() { - err := consumerGroup2.Consume(ctx2, []string{topicName}, handler2) - if err != nil && err != context.DeadlineExceeded { - t.Logf("Consumer2 error: %v", err) - } - }() + // Start both consumers with loops so they survive rebalance session exits. + go runConsumeLoop(t, ctx1, "Consumer1", consumerGroup1, []string{topicName}, handler1) + go runConsumeLoop(t, ctx2, "Consumer2", consumerGroup2, []string{topicName}, handler2) // Wait for both consumers to be ready <-handler1.ready @@ -315,7 +314,7 @@ func testMultipleConsumersJoin(t *testing.T, addr, topicName, groupID string) { handlers[i] = &RebalanceTestHandler{ messages: make(chan *sarama.ConsumerMessage, 20), ready: make(chan bool), - assignments: make(chan []int32, 5), + assignments: make(chan []int32, 1), t: t, name: fmt.Sprintf("Consumer%d", i), } @@ -323,10 +322,8 @@ func testMultipleConsumersJoin(t *testing.T, addr, topicName, groupID string) { contexts[i], cancels[i] = context.WithTimeout(context.Background(), 45*time.Second) go func(idx int) { - err := consumers[idx].Consume(contexts[idx], []string{topicName}, handlers[idx]) - if err != nil && err != context.DeadlineExceeded { - t.Logf("Consumer%d error: %v", idx, err) - } + runConsumeLoop(t, contexts[idx], fmt.Sprintf("Consumer%d", idx), + consumers[idx], []string{topicName}, handlers[idx]) }(i) } @@ -339,26 +336,43 @@ func testMultipleConsumersJoin(t *testing.T, addr, topicName, groupID string) { } }() - // Wait for all consumers to be ready + // Wait for all consumers to be ready. Four consumers racing in produces + // several rebalance rounds; each round costs ~HeartbeatInterval (3s) before + // the prior leader sees REBALANCE_IN_PROGRESS and rejoins. Give enough + // headroom for the group to converge. for i := 0; i < numConsumers; i++ { select { case <-handlers[i].ready: t.Logf("Consumer%d ready", i) - case <-time.After(15 * time.Second): + case <-time.After(30 * time.Second): t.Fatalf("Timeout waiting for Consumer%d to be ready", i) } } - // Collect final assignments from all consumers + // Collect the settled assignment for each consumer. Because each consumer's + // Consume() returns and re-enters on every rebalance, Setup fires multiple + // times while the group converges and the handler's assignments channel + // accumulates several snapshots. Drain it and keep the most recent value. assignments := make([][]int32, numConsumers) + deadline := time.Now().Add(20 * time.Second) for i := 0; i < numConsumers; i++ { select { case partitions := <-handlers[i].assignments: assignments[i] = partitions - t.Logf("Consumer%d final assignment: %v", i, partitions) - case <-time.After(20 * time.Second): + case <-time.After(time.Until(deadline)): t.Errorf("Timeout waiting for Consumer%d assignment", i) + continue } + drain: + for { + select { + case partitions := <-handlers[i].assignments: + assignments[i] = partitions + case <-time.After(3 * time.Second): + break drain + } + } + t.Logf("Consumer%d final assignment: %v", i, assignments[i]) } // Verify all partitions are assigned exactly once @@ -418,13 +432,20 @@ func (h *RebalanceTestHandler) Setup(session sarama.ConsumerGroupSession) error } } - select { - case h.assignments <- partitions: - default: - // Channel might be full, that's ok + // Always publish the latest assignment. If the channel is full, drain one + // stale value first so readers see the most recent rebalance outcome (not + // whatever arrived first while the group was still converging). + for { + select { + case h.assignments <- partitions: + return nil + default: + select { + case <-h.assignments: + default: + } + } } - - return nil } func (h *RebalanceTestHandler) Cleanup(sarama.ConsumerGroupSession) error { diff --git a/weed/mq/kafka/consumer/group_coordinator.go b/weed/mq/kafka/consumer/group_coordinator.go index 1158f9431..3bccc5da0 100644 --- a/weed/mq/kafka/consumer/group_coordinator.go +++ b/weed/mq/kafka/consumer/group_coordinator.go @@ -63,7 +63,8 @@ func (ms MemberState) String() string { // GroupMember represents a consumer in a consumer group type GroupMember struct { ID string // Member ID (generated by gateway) - ClientID string // Client ID from consumer + ClientID string // Client ID from consumer (raw Kafka header value; not unique across connections) + ConnectionID string // TCP connection identifier; distinguishes consumers that share a ClientID ClientHost string // Client host/IP GroupInstanceID *string // Static membership instance ID (optional) SessionTimeout int32 // Session timeout in milliseconds diff --git a/weed/mq/kafka/protocol/consumer_coordination.go b/weed/mq/kafka/protocol/consumer_coordination.go index dafc8c033..9e3db02aa 100644 --- a/weed/mq/kafka/protocol/consumer_coordination.go +++ b/weed/mq/kafka/protocol/consumer_coordination.go @@ -89,9 +89,17 @@ func (h *Handler) handleHeartbeat(correlationID uint32, apiVersion uint16, reque return h.buildHeartbeatErrorResponseV(correlationID, ErrorCodeUnknownMemberID, apiVersion), nil } - // Validate generation + // Validate generation. If the group is rebalancing we want Sarama to rejoin + // (via ErrRebalanceInProgress -> heartbeatLoop cancels session -> newSession) + // rather than tear down on ErrIllegalGeneration, which would exit the caller's + // Consume() entirely since Consume does not internally re-enter on that error. if request.GenerationID != group.Generation { - return h.buildHeartbeatErrorResponseV(correlationID, ErrorCodeIllegalGeneration, apiVersion), nil + switch group.State { + case consumer.GroupStatePreparingRebalance, consumer.GroupStateCompletingRebalance: + return h.buildHeartbeatErrorResponseV(correlationID, ErrorCodeRebalanceInProgress, apiVersion), nil + default: + return h.buildHeartbeatErrorResponseV(correlationID, ErrorCodeIllegalGeneration, apiVersion), nil + } } // Update member's last heartbeat @@ -124,13 +132,16 @@ func (h *Handler) handleHeartbeat(correlationID uint32, apiVersion uint16, reque func (h *Handler) handleLeaveGroup(correlationID uint32, apiVersion uint16, requestBody []byte) ([]byte, error) { // Parse LeaveGroup request - request, err := h.parseLeaveGroupRequest(requestBody) + request, err := h.parseLeaveGroupRequest(requestBody, apiVersion) if err != nil { return h.buildLeaveGroupErrorResponse(correlationID, ErrorCodeInvalidGroupID, apiVersion), nil } - // Validate request - if request.GroupID == "" || request.MemberID == "" { + // Validate request - v3+ carries member IDs inside the Members array + if request.GroupID == "" { + return h.buildLeaveGroupErrorResponse(correlationID, ErrorCodeInvalidGroupID, apiVersion), nil + } + if request.MemberID == "" && len(request.Members) == 0 { return h.buildLeaveGroupErrorResponse(correlationID, ErrorCodeInvalidGroupID, apiVersion), nil } @@ -146,66 +157,84 @@ func (h *Handler) handleLeaveGroup(correlationID uint32, apiVersion uint16, requ // Update group's last activity group.LastActivity = time.Now() - // Validate member exists - member, exists := group.Members[request.MemberID] - if !exists { - return h.buildLeaveGroupErrorResponse(correlationID, ErrorCodeUnknownMemberID, apiVersion), nil + // Normalize to a list of leaving members. v0-v2 carry a single MemberID at + // the top level; v3+ carry an array. parseLeaveGroupRequest already copies + // Members[0] to MemberID for back-compat, but the full list must be honored + // so batch departures (KIP-345) don't leave stale phantoms in the group. + leavingMembers := request.Members + if len(leavingMembers) == 0 { + leavingMembers = []LeaveGroupMember{{ + MemberID: request.MemberID, + GroupInstanceID: request.GroupInstanceID, + }} } - // For static members, only remove if GroupInstanceID matches or is not provided - if h.groupCoordinator.IsStaticMember(member) { - if request.GroupInstanceID != "" && *member.GroupInstanceID != request.GroupInstanceID { - return h.buildLeaveGroupErrorResponse(correlationID, ErrorCodeFencedInstanceID, apiVersion), nil + memberResponses := make([]LeaveGroupMemberResponse, 0, len(leavingMembers)) + removedAny := false + leaderLeft := false + + for _, leaving := range leavingMembers { + member, exists := group.Members[leaving.MemberID] + if !exists { + memberResponses = append(memberResponses, LeaveGroupMemberResponse{ + MemberID: leaving.MemberID, + GroupInstanceID: leaving.GroupInstanceID, + ErrorCode: ErrorCodeUnknownMemberID, + }) + continue } - // Unregister static member - h.groupCoordinator.UnregisterStaticMemberLocked(group, *member.GroupInstanceID) + + if h.groupCoordinator.IsStaticMember(member) { + if leaving.GroupInstanceID != "" && *member.GroupInstanceID != leaving.GroupInstanceID { + memberResponses = append(memberResponses, LeaveGroupMemberResponse{ + MemberID: leaving.MemberID, + GroupInstanceID: leaving.GroupInstanceID, + ErrorCode: ErrorCodeFencedInstanceID, + }) + continue + } + h.groupCoordinator.UnregisterStaticMemberLocked(group, *member.GroupInstanceID) + } + + if group.Leader == leaving.MemberID { + leaderLeft = true + } + delete(group.Members, leaving.MemberID) + removedAny = true + memberResponses = append(memberResponses, LeaveGroupMemberResponse{ + MemberID: leaving.MemberID, + GroupInstanceID: leaving.GroupInstanceID, + ErrorCode: ErrorCodeNone, + }) } - // Remove the member from the group - delete(group.Members, request.MemberID) - - // Update group state based on remaining members - if len(group.Members) == 0 { - // Group becomes empty - group.State = consumer.GroupStateEmpty - group.Generation++ - group.Leader = "" - } else { - // Trigger rebalancing for remaining members - group.State = consumer.GroupStatePreparingRebalance - group.Generation++ - - // If the leaving member was the leader, select a new leader - if group.Leader == request.MemberID { - // Select first remaining member as new leader - for memberID := range group.Members { - group.Leader = memberID - break + if removedAny { + // One group-state transition applied after processing all departures. + if len(group.Members) == 0 { + group.State = consumer.GroupStateEmpty + group.Generation++ + group.Leader = "" + } else { + group.State = consumer.GroupStatePreparingRebalance + group.Generation++ + if leaderLeft { + for memberID := range group.Members { + group.Leader = memberID + break + } + } + for _, m := range group.Members { + m.State = consumer.MemberStatePending } } - - // Mark remaining members as pending to trigger rebalancing - for _, member := range group.Members { - member.State = consumer.MemberStatePending - } + h.updateGroupSubscriptionFromMembers(group) } - // Update group's subscribed topics (may have changed with member leaving) - h.updateGroupSubscriptionFromMembers(group) - - // Build successful response response := LeaveGroupResponse{ CorrelationID: correlationID, ErrorCode: ErrorCodeNone, - Members: []LeaveGroupMemberResponse{ - { - MemberID: request.MemberID, - GroupInstanceID: request.GroupInstanceID, - ErrorCode: ErrorCodeNone, - }, - }, + Members: memberResponses, } - return h.buildLeaveGroupResponse(response, apiVersion), nil } @@ -329,50 +358,165 @@ func (h *Handler) parseHeartbeatRequest(data []byte, apiVersion uint16) (*Heartb }, nil } -func (h *Handler) parseLeaveGroupRequest(data []byte) (*LeaveGroupRequest, error) { +func (h *Handler) parseLeaveGroupRequest(data []byte, apiVersion uint16) (*LeaveGroupRequest, error) { if len(data) < 4 { return nil, fmt.Errorf("request too short") } offset := 0 + isFlexible := IsFlexibleVersion(uint16(APIKeyLeaveGroup), apiVersion) - // GroupID (string) - groupIDLength := int(binary.BigEndian.Uint16(data[offset:])) - offset += 2 - if offset+groupIDLength > len(data) { - return nil, fmt.Errorf("invalid group ID length") - } - groupID := string(data[offset : offset+groupIDLength]) - offset += groupIDLength - - // MemberID (string) - if offset+2 > len(data) { - return nil, fmt.Errorf("missing member ID length") - } - memberIDLength := int(binary.BigEndian.Uint16(data[offset:])) - offset += 2 - if offset+memberIDLength > len(data) { - return nil, fmt.Errorf("invalid member ID length") - } - memberID := string(data[offset : offset+memberIDLength]) - offset += memberIDLength - - // GroupInstanceID (string, v3+) - optional field - var groupInstanceID string - if offset+2 <= len(data) { - instanceIDLength := int(binary.BigEndian.Uint16(data[offset:])) - offset += 2 - if instanceIDLength != 0xFFFF && offset+instanceIDLength <= len(data) { - groupInstanceID = string(data[offset : offset+instanceIDLength]) + if isFlexible { + // Skip top-level tagged fields + _, consumed, err := DecodeTaggedFields(data[offset:]) + if err != nil { + return nil, fmt.Errorf("LeaveGroup v%d: top-level tagged fields: %w", apiVersion, err) } + offset += consumed } - return &LeaveGroupRequest{ - GroupID: groupID, - MemberID: memberID, - GroupInstanceID: groupInstanceID, - Members: []LeaveGroupMember{}, // Would parse members array for batch operations - }, nil + // GroupID + var groupID string + if isFlexible { + bytes, consumed := parseCompactString(data[offset:]) + if consumed == 0 { + return nil, fmt.Errorf("LeaveGroup v%d: invalid group ID compact string", apiVersion) + } + if bytes != nil { + groupID = string(bytes) + } + offset += consumed + } else { + if offset+2 > len(data) { + return nil, fmt.Errorf("missing group ID length") + } + groupIDLength := int(binary.BigEndian.Uint16(data[offset:])) + offset += 2 + if offset+groupIDLength > len(data) { + return nil, fmt.Errorf("invalid group ID length") + } + groupID = string(data[offset : offset+groupIDLength]) + offset += groupIDLength + } + + req := &LeaveGroupRequest{ + GroupID: groupID, + Members: []LeaveGroupMember{}, + } + + // v0-v2: top-level MemberID (string). v3+: Members array. + if apiVersion <= 2 { + if offset+2 > len(data) { + return nil, fmt.Errorf("missing member ID length") + } + memberIDLength := int(binary.BigEndian.Uint16(data[offset:])) + offset += 2 + if offset+memberIDLength > len(data) { + return nil, fmt.Errorf("invalid member ID length") + } + req.MemberID = string(data[offset : offset+memberIDLength]) + offset += memberIDLength + return req, nil + } + + // v3+: Members array [{MemberID, GroupInstanceID[, Reason]}] + var membersCount int + if isFlexible { + compactLen, consumed, err := DecodeCompactArrayLength(data[offset:]) + if err != nil { + return nil, fmt.Errorf("LeaveGroup v%d: invalid members compact array: %w", apiVersion, err) + } + membersCount = int(compactLen) + offset += consumed + } else { + if offset+4 > len(data) { + return nil, fmt.Errorf("missing members array length") + } + membersCount = int(int32(binary.BigEndian.Uint32(data[offset:]))) + offset += 4 + } + + if membersCount < 0 || membersCount > 100000 { + return nil, fmt.Errorf("unreasonable members count: %d", membersCount) + } + + for i := 0; i < membersCount; i++ { + var m LeaveGroupMember + // MemberID + if isFlexible { + bytes, consumed := parseCompactString(data[offset:]) + if consumed == 0 { + return nil, fmt.Errorf("LeaveGroup v%d: invalid members[%d].member_id", apiVersion, i) + } + if bytes != nil { + m.MemberID = string(bytes) + } + offset += consumed + } else { + if offset+2 > len(data) { + return nil, fmt.Errorf("members[%d]: missing member_id length", i) + } + ml := int(binary.BigEndian.Uint16(data[offset:])) + offset += 2 + if offset+ml > len(data) { + return nil, fmt.Errorf("members[%d]: invalid member_id length", i) + } + m.MemberID = string(data[offset : offset+ml]) + offset += ml + } + // GroupInstanceID (nullable) + if isFlexible { + bytes, consumed := parseCompactString(data[offset:]) + if consumed == 0 { + return nil, fmt.Errorf("LeaveGroup v%d: invalid members[%d].group_instance_id", apiVersion, i) + } + if bytes != nil { + m.GroupInstanceID = string(bytes) + } + offset += consumed + } else { + if offset+2 > len(data) { + return nil, fmt.Errorf("members[%d]: missing group_instance_id length", i) + } + gil := int16(binary.BigEndian.Uint16(data[offset:])) + offset += 2 + if gil >= 0 { + if offset+int(gil) > len(data) { + return nil, fmt.Errorf("members[%d]: invalid group_instance_id length", i) + } + m.GroupInstanceID = string(data[offset : offset+int(gil)]) + offset += int(gil) + } + } + // v5+: Reason (compact nullable string, flexible only) + if apiVersion >= 5 && isFlexible { + bytes, consumed := parseCompactString(data[offset:]) + if consumed > 0 { + if bytes != nil { + m.Reason = string(bytes) + } + offset += consumed + } + } + // Per-member tagged fields + if isFlexible { + _, consumed, err := DecodeTaggedFields(data[offset:]) + if err != nil { + return nil, fmt.Errorf("LeaveGroup v%d: members[%d] tagged fields: %w", apiVersion, i, err) + } + offset += consumed + } + req.Members = append(req.Members, m) + } + + // Back-compat convenience: expose the first member as top-level MemberID so + // handler code that pre-dates Members support keeps working. + if req.MemberID == "" && len(req.Members) > 0 { + req.MemberID = req.Members[0].MemberID + req.GroupInstanceID = req.Members[0].GroupInstanceID + } + + return req, nil } func (h *Handler) buildHeartbeatResponse(response HeartbeatResponse) []byte { @@ -443,10 +587,69 @@ func (h *Handler) buildLeaveGroupResponse(response LeaveGroupResponse, apiVersio return h.buildLeaveGroupV0Response(response) } - // For v1+ use the full response format + // v4+ uses flexible encoding (compact strings/arrays + tagged fields). + if IsFlexibleVersion(uint16(APIKeyLeaveGroup), apiVersion) { + return h.buildLeaveGroupFlexibleResponse(response) + } + + // v1-v3 use the full non-flexible response format. return h.buildLeaveGroupFullResponse(response) } +// buildLeaveGroupFlexibleResponse encodes a LeaveGroup response for v4+ clients, +// which negotiate flexible versions. Fields: +// +// throttle_time_ms (int32) +// error_code (int16) +// members (compact_array of {member_id: compact_string, +// group_instance_id: compact_nullable_string, +// error_code: int16, _tagged_fields}) +// _tagged_fields +func (h *Handler) buildLeaveGroupFlexibleResponse(response LeaveGroupResponse) []byte { + estimatedSize := 16 + for _, m := range response.Members { + estimatedSize += len(m.MemberID) + len(m.GroupInstanceID) + 10 + } + result := make([]byte, 0, estimatedSize) + + // throttle_time_ms + result = append(result, 0, 0, 0, 0) + + // error_code + errorCodeBytes := make([]byte, 2) + binary.BigEndian.PutUint16(errorCodeBytes, uint16(response.ErrorCode)) + result = append(result, errorCodeBytes...) + + // members: compact array + result = append(result, CompactArrayLength(uint32(len(response.Members)))...) + for _, m := range response.Members { + // member_id: compact string + result = append(result, CompactStringLength(len(m.MemberID))...) + result = append(result, []byte(m.MemberID)...) + + // group_instance_id: compact nullable string + if m.GroupInstanceID == "" { + result = append(result, 0) // null + } else { + result = append(result, CompactStringLength(len(m.GroupInstanceID))...) + result = append(result, []byte(m.GroupInstanceID)...) + } + + // member error_code + memberErrorBytes := make([]byte, 2) + binary.BigEndian.PutUint16(memberErrorBytes, uint16(m.ErrorCode)) + result = append(result, memberErrorBytes...) + + // per-member _tagged_fields (empty) + result = append(result, 0) + } + + // response-level _tagged_fields (empty) + result = append(result, 0) + + return result +} + func (h *Handler) buildLeaveGroupV0Response(response LeaveGroupResponse) []byte { result := make([]byte, 0, 6) diff --git a/weed/mq/kafka/protocol/joingroup.go b/weed/mq/kafka/protocol/joingroup.go index 85a632070..7a939ea19 100644 --- a/weed/mq/kafka/protocol/joingroup.go +++ b/weed/mq/kafka/protocol/joingroup.go @@ -83,14 +83,15 @@ func (h *Handler) handleJoinGroup(connContext *ConnectionContext, correlationID var isNewMember bool var existingMember *consumer.GroupMember - // Use the actual ClientID from Kafka protocol header for unique member ID generation - clientKey := connContext.ClientID - if clientKey == "" { - // Fallback to deterministic key if ClientID not available - clientKey = fmt.Sprintf("%s-%d-%s", request.GroupID, request.SessionTimeout, request.ProtocolType) - glog.Warningf("[JoinGroup] No ClientID in ConnectionContext for group %s, using fallback: %s", request.GroupID, clientKey) + // Dedup and memberID generation must be scoped per-connection, not per-ClientID: + // two Sarama consumers in the same process share the default ClientID ("sarama"), + // so keying on ClientID alone collapses them into a single member and breaks rebalancing. + connectionKey := connContext.ConnectionID + if connectionKey == "" { + connectionKey = fmt.Sprintf("%s-%d-%s", request.GroupID, request.SessionTimeout, request.ProtocolType) + glog.Warningf("[JoinGroup] No ConnectionID in ConnectionContext for group %s, using fallback: %s", request.GroupID, connectionKey) } else { - glog.V(1).Infof("[JoinGroup] Using ClientID from ConnectionContext for group %s: %s", request.GroupID, clientKey) + glog.V(1).Infof("[JoinGroup] Using ConnectionID %s (ClientID=%q) for group %s", connectionKey, connContext.ClientID, request.GroupID) } // Check for static membership first @@ -107,22 +108,23 @@ func (h *Handler) handleJoinGroup(connContext *ConnectionContext, correlationID } else { // Dynamic membership logic if request.MemberID == "" { - // New member - check if we already have a member for this client + // New member - reuse the existing slot only if the SAME TCP connection + // is rejoining with an empty MemberID (e.g. retry after the gateway + // evicted it). Matching on ClientID would collapse distinct consumers + // that share a default ClientID. var existingMemberID string for existingID, member := range group.Members { - if member.ClientID == clientKey && !h.groupCoordinator.IsStaticMember(member) { + if member.ConnectionID == connectionKey && !h.groupCoordinator.IsStaticMember(member) { existingMemberID = existingID break } } if existingMemberID != "" { - // Reuse existing member ID for this client memberID = existingMemberID isNewMember = false } else { - // Generate new deterministic member ID - memberID = h.groupCoordinator.GenerateMemberID(clientKey, "consumer") + memberID = h.groupCoordinator.GenerateMemberID(connectionKey, "consumer") isNewMember = true } } else { @@ -164,24 +166,36 @@ func (h *Handler) handleJoinGroup(connContext *ConnectionContext, correlationID groupInstanceID = &request.GroupInstanceID } - member := &consumer.GroupMember{ - ID: memberID, - ClientID: clientKey, // Use actual Kafka ClientID for unique member identification - ClientHost: clientHost, // Now extracted from actual connection - GroupInstanceID: groupInstanceID, - SessionTimeout: request.SessionTimeout, - RebalanceTimeout: request.RebalanceTimeout, - Subscription: h.extractSubscriptionFromProtocolsEnhanced(request.GroupProtocols), - State: consumer.MemberStatePending, - LastHeartbeat: time.Now(), - JoinedAt: time.Now(), - } - - // Add or update the member in the group before computing subscriptions or leader if group.Members == nil { group.Members = make(map[string]*consumer.GroupMember) } - group.Members[memberID] = member + + // Update existing member in place rather than replacing it, so the assignment + // (and any other state the leader populated in a prior SyncGroup) is preserved + // across rejoins. Replacing the struct clobbered member.Assignment, so the next + // non-leader SyncGroup returned an empty assignment and the consumer re-looped. + member, existing := group.Members[memberID] + // Snapshot pre-mutation state so we can roll back if the join is rejected + // (e.g. INCONSISTENT_GROUP_PROTOCOL) without corrupting an existing member. + var previousMember consumer.GroupMember + if existing { + previousMember = *member + } else { + member = &consumer.GroupMember{ + ID: memberID, + JoinedAt: time.Now(), + } + group.Members[memberID] = member + } + member.ClientID = connContext.ClientID + member.ConnectionID = connectionKey + member.ClientHost = clientHost + member.GroupInstanceID = groupInstanceID + member.SessionTimeout = request.SessionTimeout + member.RebalanceTimeout = request.RebalanceTimeout + member.Subscription = h.extractSubscriptionFromProtocolsEnhanced(request.GroupProtocols) + member.State = consumer.MemberStatePending + member.LastHeartbeat = time.Now() // Store consumer group and member ID in connection context for use in fetch requests connContext.ConsumerGroup = request.GroupID @@ -242,12 +256,20 @@ func (h *Handler) handleJoinGroup(connContext *ConnectionContext, correlationID // If a protocol is already selected for the group, reject joins that do not support it. if len(existingProtocols) > 0 && (groupProtocol == "" || groupProtocol != group.Protocol) { - // Rollback member addition and static registration before returning error - delete(group.Members, memberID) - if member.GroupInstanceID != nil && *member.GroupInstanceID != "" { - h.groupCoordinator.UnregisterStaticMemberLocked(group, *member.GroupInstanceID) + if existing { + // Existing member rejoined with an incompatible protocol — restore the + // pre-mutation snapshot rather than deleting, so we don't corrupt the + // group state for a member that was already part of it. + restored := previousMember + group.Members[memberID] = &restored + } else { + // New member was never actually part of the group: drop and unregister. + delete(group.Members, memberID) + if member.GroupInstanceID != nil && *member.GroupInstanceID != "" { + h.groupCoordinator.UnregisterStaticMemberLocked(group, *member.GroupInstanceID) + } } - // Recompute group subscription without the rejected member + // Recompute group subscription with the member state restored to pre-join. h.updateGroupSubscription(group) return h.buildJoinGroupErrorResponse(correlationID, ErrorCodeInconsistentGroupProtocol, apiVersion), nil } @@ -885,12 +907,22 @@ func (h *Handler) handleSyncGroup(correlationID uint32, apiVersion uint16, reque } glog.V(2).Infof("[SYNCGROUP] Leader assignments processed successfully, group now STABLE") } else if request.MemberID != group.Leader && len(request.GroupAssignments) == 0 { - // Non-leader member requesting its assignment - // CRITICAL FIX: Non-leader members should ALWAYS wait for leader's client-side assignments - // This is the correct behavior for Sarama and other client-side assignment protocols - glog.V(3).Infof("[SYNCGROUP] Non-leader %s waiting for/retrieving assignment in group %s (state=%s)", - request.MemberID, request.GroupID, group.State) - // Assignment will be retrieved from member.Assignment below + // Non-leader member requesting its assignment. Real Kafka blocks this + // call until the leader's SyncGroup arrives; the gateway doesn't hold the + // request, so if the leader hasn't published assignments for the current + // generation we return REBALANCE_IN_PROGRESS and let Sarama back off and + // retry the join/sync cycle (Consumer.Group.Rebalance.Retry.Max=4, + // Backoff=2s by default). If the member already has a non-empty + // assignment (steady-state re-sync) or the group has reached Stable + // (e.g. fewer partitions than members, so empty is a valid answer), + // serve member.Assignment directly. + if group.State != consumer.GroupStateStable && len(member.Assignment) == 0 { + glog.V(2).Infof("[SYNCGROUP] Non-leader %s in group %s: leader not ready (state=%s), returning REBALANCE_IN_PROGRESS", + request.MemberID, request.GroupID, group.State) + return h.buildSyncGroupErrorResponse(correlationID, ErrorCodeRebalanceInProgress, apiVersion), nil + } + glog.V(3).Infof("[SYNCGROUP] Non-leader %s retrieving assignment in group %s (state=%s assignment_len=%d)", + request.MemberID, request.GroupID, group.State, len(member.Assignment)) } else { // Trigger partition assignment using built-in strategy (server-side assignment) // This should only happen for server-side assignment protocols (not Sarama's client-side)