From ea7a41cbf11fcd3485f0e024a7d2dab92fa44f06 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Thu, 27 Aug 2026 22:25:04 +0000 Subject: [PATCH] fix(messagequeue): clarify canceled worker polls Summary: Intent: - Distinguish expected partition-worker teardown from actionable polling failures. Changes: - Log canceled in-flight polls at info only when the worker context is also canceled. - Keep active-worker cancellations and other storage failures at error level. --- .../extension/messagequeue/mysql/BUILD.bazel | 2 + .../messagequeue/mysql/subscriber.go | 13 +++ .../messagequeue/mysql/subscriber_test.go | 107 ++++++++++++++++++ 3 files changed, 122 insertions(+) diff --git a/platform/extension/messagequeue/mysql/BUILD.bazel b/platform/extension/messagequeue/mysql/BUILD.bazel index 224f63de3..12b2f7a95 100644 --- a/platform/extension/messagequeue/mysql/BUILD.bazel +++ b/platform/extension/messagequeue/mysql/BUILD.bazel @@ -52,6 +52,8 @@ go_test( "@com_github_stretchr_testify//require:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_mock//gomock:go_default_library", + "@org_uber_go_zap//:go_default_library", "@org_uber_go_zap//zaptest:go_default_library", + "@org_uber_go_zap//zaptest/observer:go_default_library", ], ) diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index 82c52b817..6e655c07f 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/platform/extension/messagequeue/mysql/subscriber.go @@ -16,6 +16,7 @@ package mysql import ( "context" + "errors" "fmt" "sort" "strconv" @@ -1014,6 +1015,18 @@ func (w *partitionWorker) run(ctx context.Context) { // return to — the only recovery is to retry on the next tick, which // happens automatically. All pollAndDeliver operations are idempotent. if err := w.pollAndDeliver(ctx); err != nil { + // A stopped worker has no next tick to recover on. Its context is + // deliberately cancelled to interrupt any in-flight store call, so + // the resulting error is part of normal teardown. + if errors.Is(err, context.Canceled) && errors.Is(ctx.Err(), context.Canceled) { + w.subscriber.logger.Infow("poll canceled while stopping partition worker", + "topic", w.sub.topic, + "partition_key", w.partitionKey, + "consumer_group", w.sub.config.ConsumerGroup, + "subscriber_name", w.sub.config.SubscriberName, + ) + return + } w.subscriber.logger.Errorw("poll failed", "topic", w.sub.topic, "partition_key", w.partitionKey, diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index 2a21856e5..91f680f09 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/platform/extension/messagequeue/mysql/subscriber_test.go @@ -25,7 +25,9 @@ import ( "github.com/stretchr/testify/require" "github.com/uber-go/tally" "go.uber.org/mock/gomock" + "go.uber.org/zap" "go.uber.org/zap/zaptest" + "go.uber.org/zap/zaptest/observer" "github.com/uber/submitqueue/platform/base/failure" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" @@ -1013,6 +1015,111 @@ func TestSubscriber_StopAllWorkers(t *testing.T) { } } +func TestPartitionWorker_RunPollErrorLogging(t *testing.T) { + tests := []struct { + name string + pollError func(context.Context) error + cancelDuringPoll bool + expectedLogError string + }{ + { + name: "worker cancellation is informational", + pollError: func(ctx context.Context) error { + <-ctx.Done() + return ctx.Err() + }, + cancelDuringPoll: true, + }, + { + name: "store cancellation on active worker is logged", + pollError: func(context.Context) error { + return context.Canceled + }, + expectedLogError: "get acked offset: context canceled", + }, + { + name: "store failure is logged", + pollError: func(context.Context) error { + return errors.New("store unavailable") + }, + expectedLogError: "get acked offset: store unavailable", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + mockMessageStore := NewMockmessageStore(ctrl) + mockOffsetStore := NewMockoffsetStore(ctrl) + mockLeaseStore := NewMockpartitionLeaseStore(ctrl) + + pollStarted := make(chan struct{}, 1) + mockOffsetStore.EXPECT().Initialize(gomock.Any(), "test_topic", "part-1", "test-consumer").Return(nil) + mockOffsetStore.EXPECT().GetAckedOffset(gomock.Any(), "test_topic", "part-1", "test-consumer").DoAndReturn( + func(ctx context.Context, _, _, _ string) (int64, error) { + select { + case pollStarted <- struct{}{}: + default: + } + return 0, tt.pollError(ctx) + }, + ).AnyTimes() + + core, logs := observer.New(zap.InfoLevel) + s := NewSubscriber( + zap.New(core).Sugar(), + tally.NoopScope, + mockMessageStore, + mockOffsetStore, + mockLeaseStore, + newTestHeartbeatStore(ctrl), + newTestDeliveryStateStore(ctrl), + ) + s.OnSignal = make(chan HookSignal, 1) + cfg := testSubscriptionConfig() + cfg.PollIntervalMs = 1 + sub := &subscription{ + topic: "test_topic", + config: cfg, + deliveryCh: make(chan extqueue.Delivery, 1), + } + worker := &partitionWorker{ + partitionKey: "part-1", + sub: sub, + subscriber: s, + done: make(chan struct{}), + } + + ctx, cancel := context.WithCancel(context.Background()) + sub.workerWg.Add(1) + go worker.run(ctx) + + <-pollStarted + if tt.cancelDuringPoll { + cancel() + } else { + <-s.OnSignal + cancel() + } + <-worker.done + + pollFailures := logs.FilterMessage("poll failed").All() + stoppedPolls := logs.FilterMessage("poll canceled while stopping partition worker").All() + if tt.expectedLogError != "" { + require.NotEmpty(t, pollFailures) + for _, entry := range pollFailures { + assert.Equal(t, zap.ErrorLevel, entry.Level) + assert.Equal(t, tt.expectedLogError, entry.ContextMap()["error"]) + } + } else { + assert.Empty(t, pollFailures) + require.Len(t, stoppedPolls, 1) + assert.Equal(t, zap.InfoLevel, stoppedPolls[0].Level) + } + }) + } +} + func TestSubscriber_FairShareCap(t *testing.T) { tests := []struct { name string