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)