From a14cbc176bf34c5a997fea73d276dc9fea00d31d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 24 Apr 2026 15:02:07 -0700 Subject: [PATCH] debug(kafka): add restart flake diagnostics --- test/kafka/e2e/offset_management_test.go | 14 +- test/kafka/internal/testutil/clients.go | 27 +++ test/kafka/internal/testutil/gateway.go | 26 +++ weed/mq/kafka/protocol/debug_state.go | 199 ++++++++++++++++++ .../kafka/protocol/fetch_partition_reader.go | 10 +- 5 files changed, 271 insertions(+), 5 deletions(-) create mode 100644 weed/mq/kafka/protocol/debug_state.go diff --git a/test/kafka/e2e/offset_management_test.go b/test/kafka/e2e/offset_management_test.go index 0df5b9d43..438373f88 100644 --- a/test/kafka/e2e/offset_management_test.go +++ b/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 { diff --git a/test/kafka/internal/testutil/clients.go b/test/kafka/internal/testutil/clients.go index 14f4aac6b..d09724c72 100644 --- a/test/kafka/internal/testutil/clients.go +++ b/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 } diff --git a/test/kafka/internal/testutil/gateway.go b/test/kafka/internal/testutil/gateway.go index 8021abcb6..ff9f37bd9 100644 --- a/test/kafka/internal/testutil/gateway.go +++ b/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 diff --git a/weed/mq/kafka/protocol/debug_state.go b/weed/mq/kafka/protocol/debug_state.go new file mode 100644 index 000000000..7961701c4 --- /dev/null +++ b/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:]) +} diff --git a/weed/mq/kafka/protocol/fetch_partition_reader.go b/weed/mq/kafka/protocol/fetch_partition_reader.go index ba6e3c38b..91306c964 100644 --- a/weed/mq/kafka/protocol/fetch_partition_reader.go +++ b/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