From 6aa353716ac0022a73150c89cf6fc13cc58b3764 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 4 May 2026 15:29:33 -0700 Subject: [PATCH] test(kafka): snapshot consumer-group state mid-attempt for resumption flake MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The onRetry hook in TestOffsetManagement/ConsumerGroupResumption only fires after defer reader.Close(), so every dump shows the post-LeaveGroup Empty state — useless for diagnosing why the second consumer hangs in the join cycle. Add an onTick callback fired every 1.5s while the reader is still joined so we can see PreparingRebalance / CompletingRebalance churn, leader, and member assignments during the 20s attempt window. --- test/kafka/e2e/offset_management_test.go | 20 ++++++++--- test/kafka/internal/testutil/clients.go | 44 ++++++++++++++++++++---- 2 files changed, 53 insertions(+), 11 deletions(-) diff --git a/test/kafka/e2e/offset_management_test.go b/test/kafka/e2e/offset_management_test.go index 438373f88..907f53f49 100644 --- a/test/kafka/e2e/offset_management_test.go +++ b/test/kafka/e2e/offset_management_test.go @@ -101,11 +101,21 @@ func testConsumerGroupResumption(t *testing.T, gateway *testutil.GatewayTestServ // 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.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) - }) + 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) + }, + func() { + // Live snapshot taken while the reader is still joined — this is + // the only place we get to see the group's transient state + // (PreparingRebalance, leader, members) during a stuck attempt. + // The onRetry snapshot above runs after reader.Close() and only + // ever shows the post-LeaveGroup Empty state. + gateway.LogConsumerGroupSnapshot(groupID) + }, + ) if err != nil { gateway.LogConsumerGroupSnapshot(groupID) } diff --git a/test/kafka/internal/testutil/clients.go b/test/kafka/internal/testutil/clients.go index d09724c72..0ed776906 100644 --- a/test/kafka/internal/testutil/clients.go +++ b/test/kafka/internal/testutil/clients.go @@ -151,20 +151,26 @@ 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) + return k.consumeWithGroup(topicName, groupID, expectedCount, nil, nil) } -func (k *KafkaGoClient) ConsumeWithGroupDebug(topicName, groupID string, expectedCount int, onRetry func(ConsumeGroupRetryDebug)) ([]kafka.Message, error) { - return k.consumeWithGroup(topicName, groupID, expectedCount, onRetry) +// ConsumeWithGroupDebug runs ConsumeWithGroup with diagnostic hooks. +// - onRetry fires after each failed attempt; the snapshot is post-LeaveGroup +// because reader.Close() has already run by then (group will look Empty). +// - onTick fires every ~1.5s while an attempt is in flight, so the caller +// can observe live group state during the join/sync/fetch cycle — the +// only place we get to see PreparingRebalance / CompletingRebalance churn. +func (k *KafkaGoClient) ConsumeWithGroupDebug(topicName, groupID string, expectedCount int, onRetry func(ConsumeGroupRetryDebug), onTick func()) ([]kafka.Message, error) { + return k.consumeWithGroup(topicName, groupID, expectedCount, onRetry, onTick) } -func (k *KafkaGoClient) consumeWithGroup(topicName, groupID string, expectedCount int, onRetry func(ConsumeGroupRetryDebug)) ([]kafka.Message, error) { +func (k *KafkaGoClient) consumeWithGroup(topicName, groupID string, expectedCount int, onRetry func(ConsumeGroupRetryDebug), onTick func()) ([]kafka.Message, error) { k.t.Helper() const maxJoinAttempts = 5 var lastErr error for attempt := 1; attempt <= maxJoinAttempts; attempt++ { - messages, err, progressed := k.consumeWithGroupOnce(topicName, groupID, expectedCount) + messages, err, progressed := k.consumeWithGroupOnce(topicName, groupID, expectedCount, onTick) if err == nil { return messages, nil } @@ -198,7 +204,7 @@ func (k *KafkaGoClient) consumeWithGroup(topicName, groupID string, expectedCoun // consumeWithGroupOnce runs a single consume attempt. Returns the messages // fetched, any error, and whether any message was received (used to decide // whether a retry is safe). -func (k *KafkaGoClient) consumeWithGroupOnce(topicName, groupID string, expectedCount int) ([]kafka.Message, error, bool) { +func (k *KafkaGoClient) consumeWithGroupOnce(topicName, groupID string, expectedCount int, onTick func()) ([]kafka.Message, error, bool) { // Give each reader its own ClientID so restarts do not get mistaken for the // still-shutting-down reader they are replacing. dialer := &kafka.Dialer{ @@ -229,6 +235,32 @@ func (k *KafkaGoClient) consumeWithGroupOnce(topicName, groupID string, expected ctx, cancel := context.WithTimeout(context.Background(), consumerGroupAttemptTimeout) defer cancel() + // While the reader is still alive (i.e. before defer reader.Close() runs), + // periodically invoke onTick so the caller can snapshot the gateway's view + // of the group during the join/sync/fetch cycle. Stop it before reader.Close + // fires — defers run LIFO, so this defer comes after the reader.Close defer. + if onTick != nil { + tickStop := make(chan struct{}) + tickDone := make(chan struct{}) + go func() { + defer close(tickDone) + ticker := time.NewTicker(1500 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-tickStop: + return + case <-ticker.C: + onTick() + } + } + }() + defer func() { + close(tickStop) + <-tickDone + }() + } + var messages []kafka.Message for i := 0; i < expectedCount; i++ { msg, err := reader.FetchMessage(ctx)