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

@@ -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

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)

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 {

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

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)
 

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)