debug(kafka): add restart flake diagnostics

test/kafka/e2e/offset_management_test.go

@@ -39,7 +39,7 @@ func TestOffsetManagement(t *testing.T) {
 	})
 
 	t.Run("ConsumerGroupResumption", func(t *testing.T) {
-		testConsumerGroupResumption(t, addr, topic2, groupID+"2")
+		testConsumerGroupResumption(t, gateway, addr, topic2, groupID+"2")
 	})
 }
 
@@ -78,7 +78,7 @@ func testBasicOffsetCommitFetch(t *testing.T, addr, topic, groupID string) {
 	t.Logf("SUCCESS: Offset management test completed - consumed %d + %d messages", len(consumed1), len(consumed2))
 }
 
-func testConsumerGroupResumption(t *testing.T, addr, topic, groupID string) {
+func testConsumerGroupResumption(t *testing.T, gateway *testutil.GatewayTestServer, addr, topic, groupID string) {
 	client := testutil.NewKafkaGoClient(t, addr)
 	msgGen := testutil.NewMessageGenerator()
 
@@ -97,10 +97,18 @@ func testConsumerGroupResumption(t *testing.T, addr, topic, groupID string) {
 	for i, msg := range consumed1 {
 		t.Logf("  Message %d: offset=%d, partition=%d, value=%s", i, msg.Offset, msg.Partition, string(msg.Value))
 	}
+	gateway.LogConsumerGroupSnapshot(groupID)
 
 	// Simulate consumer restart by consuming remaining messages with same group ID
 	t.Logf("=== Phase 3: Second consumer (simulated restart) - consuming remaining messages with same group %s ===", groupID)
-	consumed2, err := client.ConsumeWithGroup(topic, groupID, 2)
+	consumed2, err := client.ConsumeWithGroupDebug(topic, groupID, 2, func(info testutil.ConsumeGroupRetryDebug) {
+		t.Logf("Consumer restart attempt %d/%d for group %s failed before receiving any messages: %v",
+			info.Attempt, info.MaxAttempts, info.GroupID, info.Err)
+		gateway.LogConsumerGroupSnapshot(groupID)
+	})
+	if err != nil {
+		gateway.LogConsumerGroupSnapshot(groupID)
+	}
 	testutil.AssertNoError(t, err, "Failed to consume after restart")
 	t.Logf("Second consumer consumed %d messages:", len(consumed2))
 	for i, msg := range consumed2 {

test/kafka/internal/testutil/clients.go

@@ -31,6 +31,15 @@ type SaramaClient struct {
 	t          *testing.T
 }
 
+type ConsumeGroupRetryDebug struct {
+	Attempt       int
+	MaxAttempts   int
+	Topic         string
+	GroupID       string
+	ExpectedCount int
+	Err           error
+}
+
 // NewKafkaGoClient creates a new kafka-go test client
 func NewKafkaGoClient(t *testing.T, brokerAddr string) *KafkaGoClient {
 	return &KafkaGoClient{
@@ -142,6 +151,14 @@ func (k *KafkaGoClient) ConsumeMessages(topicName string, expectedCount int) ([]
 // member's LeaveGroup / session cleanup and can surface as an i/o timeout on
 // the first FetchMessage.
 func (k *KafkaGoClient) ConsumeWithGroup(topicName, groupID string, expectedCount int) ([]kafka.Message, error) {
+	return k.consumeWithGroup(topicName, groupID, expectedCount, nil)
+}
+
+func (k *KafkaGoClient) ConsumeWithGroupDebug(topicName, groupID string, expectedCount int, onRetry func(ConsumeGroupRetryDebug)) ([]kafka.Message, error) {
+	return k.consumeWithGroup(topicName, groupID, expectedCount, onRetry)
+}
+
+func (k *KafkaGoClient) consumeWithGroup(topicName, groupID string, expectedCount int, onRetry func(ConsumeGroupRetryDebug)) ([]kafka.Message, error) {
 	k.t.Helper()
 
 	const maxJoinAttempts = 5
@@ -158,6 +175,16 @@ func (k *KafkaGoClient) ConsumeWithGroup(topicName, groupID string, expectedCoun
 		if progressed {
 			return messages, err
 		}
+		if onRetry != nil {
+			onRetry(ConsumeGroupRetryDebug{
+				Attempt:       attempt,
+				MaxAttempts:   maxJoinAttempts,
+				Topic:         topicName,
+				GroupID:       groupID,
+				ExpectedCount: expectedCount,
+				Err:           err,
+			})
+		}
 		if attempt == maxJoinAttempts {
 			break
 		}

test/kafka/internal/testutil/gateway.go

@@ -2,6 +2,7 @@ package testutil
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net"
 	"os"
@@ -111,6 +112,31 @@ func (g *GatewayTestServer) CleanupAndClose() {
 	}
 }
 
+// LogConsumerGroupSnapshot dumps the gateway's current view of a consumer group.
+func (g *GatewayTestServer) LogConsumerGroupSnapshot(groupID string) {
+	g.t.Helper()
+
+	handler := g.GetHandler()
+	if handler == nil {
+		g.t.Logf("Consumer group snapshot for %s unavailable: handler is nil", groupID)
+		return
+	}
+
+	snapshot := handler.DebugGroupSnapshot(groupID)
+	if snapshot == nil {
+		g.t.Logf("Consumer group snapshot for %s unavailable", groupID)
+		return
+	}
+
+	payload, err := json.MarshalIndent(snapshot, "", "  ")
+	if err != nil {
+		g.t.Logf("Consumer group snapshot for %s could not be marshaled: %v", groupID, err)
+		return
+	}
+
+	g.t.Logf("Consumer group snapshot for %s:\n%s", groupID, string(payload))
+}
+
 // SMQAvailabilityMode indicates whether SeaweedMQ is available for testing
 type SMQAvailabilityMode int
 

weed/mq/kafka/protocol/debug_state.go

@@ -0,0 +1,199 @@
+package protocol
+
+import (
+	"sort"
+	"time"
+)
+
+type DebugGroupSnapshot struct {
+	GroupID          string                      `json:"group_id"`
+	Exists           bool                        `json:"exists"`
+	KnownGroups      []string                    `json:"known_groups,omitempty"`
+	State            string                      `json:"state,omitempty"`
+	Generation       int32                       `json:"generation,omitempty"`
+	Protocol         string                      `json:"protocol,omitempty"`
+	Leader           string                      `json:"leader,omitempty"`
+	CreatedAt        time.Time                   `json:"created_at,omitempty"`
+	LastActivity     time.Time                   `json:"last_activity,omitempty"`
+	SubscribedTopics []string                    `json:"subscribed_topics,omitempty"`
+	Members          []DebugGroupMemberSnapshot  `json:"members,omitempty"`
+	OffsetCommits    []DebugOffsetCommitSnapshot `json:"offset_commits,omitempty"`
+}
+
+type DebugGroupMemberSnapshot struct {
+	MemberID           string    `json:"member_id"`
+	ClientID           string    `json:"client_id,omitempty"`
+	ConnectionID       string    `json:"connection_id,omitempty"`
+	ClientHost         string    `json:"client_host,omitempty"`
+	State              string    `json:"state,omitempty"`
+	SessionTimeoutMs   int32     `json:"session_timeout_ms,omitempty"`
+	RebalanceTimeoutMs int32     `json:"rebalance_timeout_ms,omitempty"`
+	JoinedAt           time.Time `json:"joined_at,omitempty"`
+	LastHeartbeat      time.Time `json:"last_heartbeat,omitempty"`
+	LastHeartbeatAgeMs int64     `json:"last_heartbeat_age_ms,omitempty"`
+	Subscription       []string  `json:"subscription,omitempty"`
+	AssignedPartitions []string  `json:"assigned_partitions,omitempty"`
+	GroupInstanceID    string    `json:"group_instance_id,omitempty"`
+}
+
+type DebugOffsetCommitSnapshot struct {
+	Topic       string    `json:"topic"`
+	Partition   int32     `json:"partition"`
+	Offset      int64     `json:"offset"`
+	Metadata    string    `json:"metadata,omitempty"`
+	Timestamp   time.Time `json:"timestamp,omitempty"`
+	CommitAgeMs int64     `json:"commit_age_ms,omitempty"`
+}
+
+func (h *Handler) DebugGroupSnapshot(groupID string) *DebugGroupSnapshot {
+	snapshot := &DebugGroupSnapshot{
+		GroupID:     groupID,
+		KnownGroups: h.debugKnownGroups(),
+	}
+	if h == nil || h.groupCoordinator == nil {
+		return snapshot
+	}
+
+	group := h.groupCoordinator.GetGroup(groupID)
+	if group == nil {
+		return snapshot
+	}
+
+	now := time.Now()
+
+	group.Mu.RLock()
+	defer group.Mu.RUnlock()
+
+	snapshot.Exists = true
+	snapshot.State = group.State.String()
+	snapshot.Generation = group.Generation
+	snapshot.Protocol = group.Protocol
+	snapshot.Leader = group.Leader
+	snapshot.CreatedAt = group.CreatedAt
+	snapshot.LastActivity = group.LastActivity
+
+	if len(group.SubscribedTopics) > 0 {
+		topics := make([]string, 0, len(group.SubscribedTopics))
+		for topic := range group.SubscribedTopics {
+			topics = append(topics, topic)
+		}
+		sort.Strings(topics)
+		snapshot.SubscribedTopics = topics
+	}
+
+	memberIDs := make([]string, 0, len(group.Members))
+	for memberID := range group.Members {
+		memberIDs = append(memberIDs, memberID)
+	}
+	sort.Strings(memberIDs)
+
+	snapshot.Members = make([]DebugGroupMemberSnapshot, 0, len(memberIDs))
+	for _, memberID := range memberIDs {
+		member := group.Members[memberID]
+		if member == nil {
+			continue
+		}
+
+		subscription := append([]string(nil), member.Subscription...)
+		sort.Strings(subscription)
+
+		assignments := make([]string, 0, len(member.Assignment))
+		for _, assignment := range member.Assignment {
+			assignments = append(assignments, assignment.Topic+":"+itoa32(assignment.Partition))
+		}
+		sort.Strings(assignments)
+
+		groupInstanceID := ""
+		if member.GroupInstanceID != nil {
+			groupInstanceID = *member.GroupInstanceID
+		}
+
+		snapshot.Members = append(snapshot.Members, DebugGroupMemberSnapshot{
+			MemberID:           member.ID,
+			ClientID:           member.ClientID,
+			ConnectionID:       member.ConnectionID,
+			ClientHost:         member.ClientHost,
+			State:              member.State.String(),
+			SessionTimeoutMs:   member.SessionTimeout,
+			RebalanceTimeoutMs: member.RebalanceTimeout,
+			JoinedAt:           member.JoinedAt,
+			LastHeartbeat:      member.LastHeartbeat,
+			LastHeartbeatAgeMs: ageMillis(now, member.LastHeartbeat),
+			Subscription:       subscription,
+			AssignedPartitions: assignments,
+			GroupInstanceID:    groupInstanceID,
+		})
+	}
+
+	if len(group.OffsetCommits) > 0 {
+		topics := make([]string, 0, len(group.OffsetCommits))
+		for topic := range group.OffsetCommits {
+			topics = append(topics, topic)
+		}
+		sort.Strings(topics)
+
+		offsets := make([]DebugOffsetCommitSnapshot, 0)
+		for _, topic := range topics {
+			partitions := make([]int32, 0, len(group.OffsetCommits[topic]))
+			for partition := range group.OffsetCommits[topic] {
+				partitions = append(partitions, partition)
+			}
+			sort.Slice(partitions, func(i, j int) bool { return partitions[i] < partitions[j] })
+			for _, partition := range partitions {
+				offsetCommit := group.OffsetCommits[topic][partition]
+				offsets = append(offsets, DebugOffsetCommitSnapshot{
+					Topic:       topic,
+					Partition:   partition,
+					Offset:      offsetCommit.Offset,
+					Metadata:    offsetCommit.Metadata,
+					Timestamp:   offsetCommit.Timestamp,
+					CommitAgeMs: ageMillis(now, offsetCommit.Timestamp),
+				})
+			}
+		}
+		snapshot.OffsetCommits = offsets
+	}
+
+	return snapshot
+}
+
+func (h *Handler) debugKnownGroups() []string {
+	if h == nil || h.groupCoordinator == nil {
+		return nil
+	}
+
+	groupIDs := h.groupCoordinator.ListGroups()
+	sort.Strings(groupIDs)
+	return groupIDs
+}
+
+func ageMillis(now, then time.Time) int64 {
+	if then.IsZero() {
+		return 0
+	}
+	return now.Sub(then).Milliseconds()
+}
+
+func itoa32(value int32) string {
+	if value == 0 {
+		return "0"
+	}
+
+	negative := value < 0
+	if negative {
+		value = -value
+	}
+
+	var digits [12]byte
+	i := len(digits)
+	for value > 0 {
+		i--
+		digits[i] = byte('0' + (value % 10))
+		value /= 10
+	}
+	if negative {
+		i--
+		digits[i] = '-'
+	}
+	return string(digits[i:])
+}

weed/mq/kafka/protocol/fetch_partition_reader.go

@@ -187,8 +187,14 @@ func (pr *partitionReader) serveFetchRequest(ctx context.Context, req *partition
 
 	// Log what we got back - DETAILED for diagnostics
 	if len(recordBatch) == 0 {
-		glog.V(2).Infof("[%s] FETCH %s[%d]: readRecords returned EMPTY (offset=%d, hwm=%d)",
-			pr.connCtx.ConnectionID, pr.topicName, pr.partitionID, req.requestedOffset, result.highWaterMark)
+		if req.requestedOffset < result.highWaterMark {
+			glog.Warningf("[%s] FETCH GAP for %s[%d]: group=%s member=%s requested offset=%d below HWM=%d but readRecords returned empty (maxWaitMs=%d)",
+				pr.connCtx.ConnectionID, pr.topicName, pr.partitionID, pr.connCtx.ConsumerGroup, pr.connCtx.MemberID,
+				req.requestedOffset, result.highWaterMark, req.maxWaitMs)
+		} else {
+			glog.V(2).Infof("[%s] FETCH %s[%d]: readRecords returned EMPTY (offset=%d, hwm=%d)",
+				pr.connCtx.ConnectionID, pr.topicName, pr.partitionID, req.requestedOffset, result.highWaterMark)
+		}
 		result.recordBatch = []byte{}
 	} else {
 		result.recordBatch = recordBatch