From c37eaea6573d6ad149f9301271530ef8e02d8aae Mon Sep 17 00:00:00 2001 From: jamy Date: Mon, 20 Jul 2026 18:53:26 +0000 Subject: [PATCH] feat(platform): add pipeline.Construct engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary - Introduces `platform/pipeline` with the typed `Construct[D]` engine - `Stage[D]` struct declares pipeline topology as a typed table (key, name, consumer group, controller constructor, optional DLQ) - Engine builds topic registry, creates primary + DLQ consumers, eagerly constructs all controllers, pairs DLQ stages automatically - Returns a `lifecycle.Component` for ordered start/stop - Options: `TopicNames`, `Classifiers`, `PublishOnly`, `ExtraComponents` - Pure addition — no existing code changes Step 2 of the [Modular Queue Wiring RFC](https://github.com/uber/submitqueue/blob/main/doc/rfc/submitqueue/modular-queue-wiring.md). Stacked on #402. ## Test plan - [x] 10 unit tests: single stage, DLQ pairing, multiple stages, empty stages error, controller creation failure, DLQ creation failure, publish-only topics, topic name overrides, resolveTopicName table test, dlqTopicKey, buildTopicConfigs - [x] `bazel test //platform/pipeline:go_default_test` passes - [x] `make gazelle` — BUILD.bazel in sync - [x] `make fmt` — code formatted Co-Authored-By: Claude Opus 4.6 --- platform/pipeline/BUILD.bazel | 32 +++ platform/pipeline/pipeline.go | 281 +++++++++++++++++++++++ platform/pipeline/pipeline_test.go | 357 +++++++++++++++++++++++++++++ 3 files changed, 670 insertions(+) create mode 100644 platform/pipeline/BUILD.bazel create mode 100644 platform/pipeline/pipeline.go create mode 100644 platform/pipeline/pipeline_test.go diff --git a/platform/pipeline/BUILD.bazel b/platform/pipeline/BUILD.bazel new file mode 100644 index 00000000..001b3468 --- /dev/null +++ b/platform/pipeline/BUILD.bazel @@ -0,0 +1,32 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["pipeline.go"], + importpath = "github.com/uber/submitqueue/platform/pipeline", + visibility = ["//visibility:public"], + deps = [ + "//platform/consumer:go_default_library", + "//platform/errs:go_default_library", + "//platform/extension/messagequeue:go_default_library", + "//platform/lifecycle:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["pipeline_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/consumer:go_default_library", + "//platform/extension/messagequeue:go_default_library", + "//platform/extension/messagequeue/mock:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@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", + ], +) diff --git a/platform/pipeline/pipeline.go b/platform/pipeline/pipeline.go new file mode 100644 index 00000000..a4ff8c5e --- /dev/null +++ b/platform/pipeline/pipeline.go @@ -0,0 +1,281 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package pipeline provides a typed engine for assembling queue-driven +// service pipelines from declarative data. A service declares its topology +// as a []Stage[D] table and its dependencies as a Deps struct; Construct +// builds all consumers, registers controllers, pairs DLQ stages, and +// returns a single lifecycle.Component the host drives with Start/Stop. +package pipeline + +import ( + "context" + "fmt" + "time" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/errs" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" + "github.com/uber/submitqueue/platform/lifecycle" + "go.uber.org/zap" +) + +// timeNow is a hook for tests to control time. Production uses time.Now. +var timeNow = time.Now + +// Stage is one row of a service's topology table. D is the service's Deps type. +type Stage[D any] struct { + // Key is the stage's logical topic key (e.g. topickey.TopicKeyStart). + // The engine maps it to a physical topic name via the TopicNames option. + Key consumer.TopicKey + + // Name is the physical topic name for this stage (e.g. "start"). + // Used as the default when no TopicNames override is provided. + Name string + + // ConsumerGroup is the consumer group suffix for this stage's subscription + // (e.g. "orchestrator-start"). + ConsumerGroup string + + // New builds the stage's controller from the service's Deps. The engine + // calls it once, eagerly, inside Construct — so a nil/missing dependency + // fails at boot with the stage's name on it, never mid-delivery. + New func(D) (consumer.Controller, error) + + // DLQ, when non-nil, declares "this stage dead-letters". The engine then + // derives the paired DLQ topic (_dlq, retry budget, DLQ-of-DLQ + // disabled) AND registers this reconciler on the DLQ consumer. Declaring + // one without getting the other is impossible — that's the invariant. + DLQ func(D) (consumer.Controller, error) +} + +// PublishOnlyTopic declares a topic the service publishes to but does not +// consume. The engine registers it in the TopicRegistry so controllers +// can publish to it, but creates no subscription or controller. +type PublishOnlyTopic struct { + // Key is the logical topic key. + Key consumer.TopicKey + + // Name is the physical topic name. + Name string +} + +// options holds the resolved configuration for a Construct call. +type options struct { + topicNames map[consumer.TopicKey]string + classifiers []errs.Classifier + publishOnly []PublishOnlyTopic + extraComponents []lifecycle.Component +} + +// Option configures a Construct call. +type Option func(*options) + +// TopicNames provides a mapping from logical topic keys to physical topic +// names. Keys not present in the map fall back to the Stage.Name default. +func TopicNames(m map[consumer.TopicKey]string) Option { + return func(o *options) { o.topicNames = m } +} + +// Classifiers sets the error classifiers for the primary consumer's +// ErrorProcessor. DLQ consumers always use AlwaysRetryableProcessor. +func Classifiers(c ...errs.Classifier) Option { + return func(o *options) { o.classifiers = c } +} + +// PublishOnly adds topics the service publishes to but does not consume. +func PublishOnly(topics ...PublishOnlyTopic) Option { + return func(o *options) { o.publishOnly = append(o.publishOnly, topics...) } +} + +// ExtraComponents adds lifecycle components that are started before +// consumers and stopped after them. +func ExtraComponents(c ...lifecycle.Component) Option { + return func(o *options) { o.extraComponents = append(o.extraComponents, c...) } +} + +// dlqTopicKey returns the DLQ topic key for a primary stage key. +// Matches the convention in submitqueue/orchestrator/controller/dlq.TopicKey. +const dlqTopicSuffix = "_dlq" + +func dlqTopicKey(primary consumer.TopicKey) consumer.TopicKey { + return consumer.TopicKey(string(primary) + dlqTopicSuffix) +} + +// Construct is the single assembly function for a queue-driven service. +// It builds the topic registry, creates primary and DLQ consumers, +// eagerly constructs all controllers, and returns a lifecycle.Component +// that starts and stops everything in the correct order. +// +// The returned Component starts in this order: +// 1. Extra components (infrastructure) +// 2. Primary consumer (work-accepting) +// 3. DLQ consumer (reconciliation) +// +// Stop reverses the order: DLQ consumer drains first, then primary, then +// infrastructure. +func Construct[D any]( + logger *zap.SugaredLogger, + scope tally.Scope, + queue extqueue.Queue, + subscriberName string, + deps D, + stages []Stage[D], + opts ...Option, +) (lifecycle.Component, error) { + if len(stages) == 0 { + return nil, fmt.Errorf("pipeline: at least one stage is required") + } + + o := &options{} + for _, opt := range opts { + opt(o) + } + + // Build topic configs for the registry. + configs, err := buildTopicConfigs(queue, subscriberName, stages, o) + if err != nil { + return nil, err + } + + registry, err := consumer.NewTopicRegistry(configs) + if err != nil { + return nil, fmt.Errorf("pipeline: failed to create topic registry: %w", err) + } + + // Create the primary consumer with user-provided classifiers. + primaryProcessor := errs.NewClassifierProcessor(o.classifiers...) + primary := consumer.New(logger, scope, registry, primaryProcessor) + + // Create the DLQ consumer with always-retryable processor. + dlq := consumer.New(logger, scope, registry, errs.AlwaysRetryableProcessor) + + hasDLQ := false + + // Eagerly construct and register all controllers. + for _, s := range stages { + ctl, err := s.New(deps) + if err != nil { + return nil, fmt.Errorf("pipeline: stage %s: failed to create controller: %w", s.Key, err) + } + if err := primary.Register(ctl); err != nil { + return nil, fmt.Errorf("pipeline: stage %s: failed to register controller: %w", s.Key, err) + } + + if s.DLQ != nil { + rec, err := s.DLQ(deps) + if err != nil { + return nil, fmt.Errorf("pipeline: stage %s dlq: failed to create controller: %w", s.Key, err) + } + if err := dlq.Register(rec); err != nil { + return nil, fmt.Errorf("pipeline: stage %s dlq: failed to register controller: %w", s.Key, err) + } + hasDLQ = true + } + } + + // Compose the lifecycle group. + members := make([]lifecycle.Component, 0, len(o.extraComponents)+2) + members = append(members, o.extraComponents...) + members = append(members, &consumerComponent{name: "primary", c: primary}) + if hasDLQ { + members = append(members, &consumerComponent{name: "dlq", c: dlq}) + } + + return lifecycle.NewGroup(members...), nil +} + +// buildTopicConfigs constructs the []consumer.TopicConfig from stages and options. +func buildTopicConfigs[D any]( + queue extqueue.Queue, + subscriberName string, + stages []Stage[D], + o *options, +) ([]consumer.TopicConfig, error) { + // Pre-size: each stage gets a primary config + optional DLQ config, + // plus publish-only topics. + configs := make([]consumer.TopicConfig, 0, 2*len(stages)+len(o.publishOnly)) + + for _, s := range stages { + topicName := resolveTopicName(s.Key, s.Name, o.topicNames) + + configs = append(configs, consumer.TopicConfig{ + Key: s.Key, + Name: topicName, + Queue: queue, + Subscription: extqueue.DefaultSubscriptionConfig( + subscriberName, s.ConsumerGroup, + ), + }) + + if s.DLQ != nil { + configs = append(configs, consumer.TopicConfig{ + Key: dlqTopicKey(s.Key), + Name: topicName + dlqTopicSuffix, + Queue: queue, + Subscription: extqueue.DLQSubscriptionConfig( + subscriberName, s.ConsumerGroup+"-dlq", + ), + }) + } + } + + for _, p := range o.publishOnly { + topicName := resolveTopicName(p.Key, p.Name, o.topicNames) + configs = append(configs, consumer.TopicConfig{ + Key: p.Key, + Name: topicName, + Queue: queue, + }) + } + + return configs, nil +} + +// resolveTopicName returns the override name if present, otherwise the default. +func resolveTopicName(key consumer.TopicKey, defaultName string, overrides map[consumer.TopicKey]string) string { + if overrides != nil { + if name, ok := overrides[key]; ok { + return name + } + } + return defaultName +} + +// consumerComponent adapts consumer.Consumer to lifecycle.Component. +// Consumer.Stop takes a timeoutMs int64; Component.Stop takes a context. +// We derive the timeout from the context's deadline if set, defaulting to 30s. +type consumerComponent struct { + name string + c consumer.Consumer +} + +func (a *consumerComponent) Start(ctx context.Context) error { + return a.c.Start(ctx) +} + +func (a *consumerComponent) Stop(ctx context.Context) error { + const defaultStopTimeoutMs = 30000 + timeoutMs := int64(defaultStopTimeoutMs) + if deadline, ok := ctx.Deadline(); ok { + remaining := deadline.Sub(timeNow()) + if remaining > 0 { + timeoutMs = remaining.Milliseconds() + } else { + timeoutMs = 0 + } + } + return a.c.Stop(timeoutMs) +} diff --git a/platform/pipeline/pipeline_test.go b/platform/pipeline/pipeline_test.go new file mode 100644 index 00000000..9280b145 --- /dev/null +++ b/platform/pipeline/pipeline_test.go @@ -0,0 +1,357 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package pipeline + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" + mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +// testDeps is a minimal Deps type for testing. +type testDeps struct { + logger *zap.SugaredLogger +} + +// fakeController satisfies consumer.Controller for testing. +type fakeController struct { + key consumer.TopicKey + group string +} + +func (f *fakeController) Process(_ context.Context, _ consumer.Delivery) error { return nil } +func (f *fakeController) Name() string { return string(f.key) } +func (f *fakeController) TopicKey() consumer.TopicKey { return f.key } +func (f *fakeController) ConsumerGroup() string { return f.group } + +func newTestLogger() *zap.SugaredLogger { + l, _ := zap.NewDevelopment() + return l.Sugar() +} + +func TestConstruct_SingleStage_NoDLQ(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps) (consumer.Controller, error) { + return &fakeController{key: "start", group: "orchestrator-start"}, nil + }, + }, + } + + comp, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages) + require.NoError(t, err) + assert.NotNil(t, comp) +} + +func TestConstruct_WithDLQ(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps) (consumer.Controller, error) { + return &fakeController{key: "start", group: "orchestrator-start"}, nil + }, + DLQ: func(d testDeps) (consumer.Controller, error) { + return &fakeController{key: "start_dlq", group: "orchestrator-start-dlq"}, nil + }, + }, + } + + comp, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages) + require.NoError(t, err) + assert.NotNil(t, comp) +} + +func TestConstruct_MultipleStages(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps) (consumer.Controller, error) { + return &fakeController{key: "start", group: "orchestrator-start"}, nil + }, + }, + { + Key: "validate", + Name: "validate", + ConsumerGroup: "orchestrator-validate", + New: func(d testDeps) (consumer.Controller, error) { + return &fakeController{key: "validate", group: "orchestrator-validate"}, nil + }, + DLQ: func(d testDeps) (consumer.Controller, error) { + return &fakeController{key: "validate_dlq", group: "orchestrator-validate-dlq"}, nil + }, + }, + } + + comp, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages) + require.NoError(t, err) + assert.NotNil(t, comp) +} + +func TestConstruct_EmptyStages_Error(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + + deps := testDeps{logger: newTestLogger()} + _, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, nil) + require.Error(t, err) + assert.Contains(t, err.Error(), "at least one stage is required") +} + +func TestConstruct_ControllerCreationFailure(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps) (consumer.Controller, error) { + return nil, fmt.Errorf("missing dependency") + }, + }, + } + + _, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages) + require.Error(t, err) + assert.Contains(t, err.Error(), "stage start") + assert.Contains(t, err.Error(), "missing dependency") +} + +func TestConstruct_DLQControllerCreationFailure(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps) (consumer.Controller, error) { + return &fakeController{key: "start", group: "orchestrator-start"}, nil + }, + DLQ: func(d testDeps) (consumer.Controller, error) { + return nil, fmt.Errorf("dlq dependency missing") + }, + }, + } + + _, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages) + require.Error(t, err) + assert.Contains(t, err.Error(), "stage start dlq") + assert.Contains(t, err.Error(), "dlq dependency missing") +} + +func TestConstruct_WithPublishOnly(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps) (consumer.Controller, error) { + return &fakeController{key: "start", group: "orchestrator-start"}, nil + }, + }, + } + + comp, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages, + PublishOnly( + PublishOnlyTopic{Key: "log", Name: "log"}, + PublishOnlyTopic{Key: "merge-request", Name: "merge-request"}, + ), + ) + require.NoError(t, err) + assert.NotNil(t, comp) +} + +func TestConstruct_WithTopicNameOverrides(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps) (consumer.Controller, error) { + return &fakeController{key: "start", group: "orchestrator-start"}, nil + }, + }, + } + + comp, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages, + TopicNames(map[consumer.TopicKey]string{ + "start": "custom-start-topic", + }), + ) + require.NoError(t, err) + assert.NotNil(t, comp) +} + +func TestResolveTopicName(t *testing.T) { + tests := []struct { + name string + key consumer.TopicKey + defaultN string + overrides map[consumer.TopicKey]string + want string + }{ + { + name: "no overrides", + key: "start", + defaultN: "start", + want: "start", + }, + { + name: "nil overrides", + key: "start", + defaultN: "start", + overrides: nil, + want: "start", + }, + { + name: "key not in overrides", + key: "start", + defaultN: "start", + overrides: map[consumer.TopicKey]string{"other": "other-name"}, + want: "start", + }, + { + name: "key in overrides", + key: "start", + defaultN: "start", + overrides: map[consumer.TopicKey]string{"start": "custom-start"}, + want: "custom-start", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := resolveTopicName(tt.key, tt.defaultN, tt.overrides) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestDLQTopicKey(t *testing.T) { + assert.Equal(t, consumer.TopicKey("start_dlq"), dlqTopicKey("start")) + assert.Equal(t, consumer.TopicKey("validate_dlq"), dlqTopicKey("validate")) +} + +func TestBuildTopicConfigs(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps) (consumer.Controller, error) { return nil, nil }, + DLQ: func(d testDeps) (consumer.Controller, error) { return nil, nil }, + }, + { + Key: "validate", + Name: "validate", + ConsumerGroup: "orchestrator-validate", + New: func(d testDeps) (consumer.Controller, error) { return nil, nil }, + // No DLQ for this stage. + }, + } + + o := &options{ + publishOnly: []PublishOnlyTopic{ + {Key: "log", Name: "log"}, + }, + } + + configs, err := buildTopicConfigs(q, "test-sub", stages, o) + require.NoError(t, err) + + // Expected: start (primary + DLQ) + validate (primary only) + log (publish-only) = 4 + assert.Len(t, configs, 4) + + // Verify primary stage config. + assert.Equal(t, consumer.TopicKey("start"), configs[0].Key) + assert.Equal(t, "start", configs[0].Name) + assert.Equal(t, "orchestrator-start", configs[0].Subscription.ConsumerGroup) + + // Verify DLQ config derived from primary. + assert.Equal(t, consumer.TopicKey("start_dlq"), configs[1].Key) + assert.Equal(t, "start_dlq", configs[1].Name) + assert.Equal(t, "orchestrator-start-dlq", configs[1].Subscription.ConsumerGroup) + + // Verify DLQ subscription has DLQ disabled (no cascade). + expected := extqueue.DLQSubscriptionConfig("test-sub", "orchestrator-start-dlq") + assert.Equal(t, expected.DLQ.Enabled, configs[1].Subscription.DLQ.Enabled) + assert.Equal(t, expected.Retry.MaxAttempts, configs[1].Subscription.Retry.MaxAttempts) + + // Verify validate stage (no DLQ). + assert.Equal(t, consumer.TopicKey("validate"), configs[2].Key) + + // Verify publish-only topic. + assert.Equal(t, consumer.TopicKey("log"), configs[3].Key) + assert.Equal(t, "log", configs[3].Name) + assert.Equal(t, "", configs[3].Subscription.ConsumerGroup) +}