Merge #152598

152598: changefeedccl: fix db-level feed error handling in changefeed processors r=andyyang890,aerfrei a=KeithCh

AllTargets can error when called from the changefeed processor due to
flakes unrelated to changefeeds. This commit fixes the error handling 
so that it doesn't result in a panic.

Part of: #147371
Epic: CRDB-1421

Release note: none

Co-authored-by: Keith Chow <[email protected]>

Author: craig[bot]

pkg/ccl/changefeedccl/changefeed.go

@@ -43,12 +43,8 @@ type ChangefeedConfig struct {
 // makeChangefeedConfigFromJobDetails creates a ChangefeedConfig struct from any
 // version of the ChangefeedDetails protobuf.
 func makeChangefeedConfigFromJobDetails(
-	ctx context.Context, d jobspb.ChangefeedDetails, execCfg *sql.ExecutorConfig,
+	d jobspb.ChangefeedDetails, targets changefeedbase.Targets,
 ) (ChangefeedConfig, error) {
-	targets, err := AllTargets(ctx, d, execCfg)
-	if err != nil {
-		return ChangefeedConfig{}, err
-	}
 	return ChangefeedConfig{
 		SinkURI:  d.SinkURI,
 		Opts:     changefeedbase.MakeStatementOptions(d.Opts),

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -201,19 +201,20 @@ func newChangeAggregatorProcessor(
 ) (_ execinfra.Processor, retErr error) {
 	// Setup monitoring for this node drain.
 	drainWatcher, drainDone := makeDrainWatcher(flowCtx)
-	defer func() {
-		if retErr != nil {
-			drainDone()
-		}
-	}()
-
 	memMonitor := execinfra.NewMonitor(ctx, flowCtx.Mon, mon.MakeName("changeagg-mem"))
 	ca := &changeAggregator{
 		spec:              spec,
 		memAcc:            memMonitor.MakeBoundAccount(),
 		checkForNodeDrain: drainWatcher.checkForNodeDrain,
 	}
 
+	defer func() {
+		if retErr != nil {
+			ca.close()
+			drainDone()
+		}
+	}()
+
 	if err := ca.Init(
 		ctx,
 		ca,
@@ -351,7 +352,19 @@ func (ca *changeAggregator) Start(ctx context.Context) {
 		return
 	}
 
-	feed, err := makeChangefeedConfigFromJobDetails(ctx, ca.spec.Feed, ca.FlowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig))
+	execCfg := ca.FlowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig)
+	if ca.knobs.OverrideExecCfg != nil {
+		execCfg = ca.knobs.OverrideExecCfg(execCfg)
+	}
+	ca.targets, err = AllTargets(ctx, ca.spec.Feed, execCfg)
+	if err != nil {
+		log.Dev.Warningf(ca.Ctx(), "moving to draining due to error getting targets: %v", err)
+		ca.MoveToDraining(err)
+		ca.cancel()
+		return
+	}
+
+	feed, err := makeChangefeedConfigFromJobDetails(ca.spec.Feed, ca.targets)
 	if err != nil {
 		log.Dev.Warningf(ca.Ctx(), "moving to draining due to error making changefeed config: %v", err)
 		ca.MoveToDraining(err)
@@ -382,13 +395,6 @@ func (ca *changeAggregator) Start(ctx context.Context) {
 		return
 	}
 	ca.sliMetricsID = ca.sliMetrics.claimId()
-	ca.targets, err = AllTargets(ctx, ca.spec.Feed, ca.FlowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig))
-	if err != nil {
-		log.Dev.Warningf(ca.Ctx(), "moving to draining due to error getting targets: %v", err)
-		ca.MoveToDraining(err)
-		ca.cancel()
-		return
-	}
 
 	recorder := metricsRecorder(ca.sliMetrics)
 	recorder, err = ca.wrapMetricsRecorderWithTelemetry(ctx, recorder, ca.targets)
@@ -1231,7 +1237,7 @@ func newChangeFrontierProcessor(
 	spec execinfrapb.ChangeFrontierSpec,
 	input execinfra.RowSource,
 	post *execinfrapb.PostProcessSpec,
-) (execinfra.Processor, error) {
+) (_ execinfra.Processor, retErr error) {
 	memMonitor := execinfra.NewMonitor(ctx, flowCtx.Mon, mon.MakeName("changefntr-mem"))
 
 	cf := &changeFrontier{
@@ -1244,6 +1250,12 @@ func newChangeFrontierProcessor(
 		usageWgCancel: func() {},
 	}
 
+	defer func() {
+		if retErr != nil {
+			cf.close()
+		}
+	}()
+
 	if cfKnobs, ok := flowCtx.TestingKnobs().Changefeed.(*TestingKnobs); ok {
 		cf.knobs = *cfKnobs
 	}
@@ -1306,7 +1318,11 @@ func newChangeFrontierProcessor(
 	if err != nil {
 		return nil, err
 	}
-	targets, err := AllTargets(ctx, spec.Feed, flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig))
+	execCfg := flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig)
+	if cf.knobs.OverrideExecCfg != nil {
+		execCfg = cf.knobs.OverrideExecCfg(execCfg)
+	}
+	targets, err := AllTargets(ctx, spec.Feed, execCfg)
 	if err != nil {
 		return nil, err
 	}

pkg/ccl/changefeedccl/event_processing_test.go

@@ -153,16 +153,16 @@ func TestTopicForEvent(t *testing.T) {
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			cfg, err := makeChangefeedConfigFromJobDetails(context.Background(), tc.details, nil)
+			// Can pass in nil for execCfg to AllTargets because we're not testing
+			// database-level targets.
+			targets, err := AllTargets(context.Background(), tc.details, nil)
+			require.NoError(t, err)
+			cfg, err := makeChangefeedConfigFromJobDetails(tc.details, targets)
 			require.NoError(t, err)
 			c := kvEventToRowConsumer{
 				details:              cfg,
 				topicDescriptorCache: make(map[TopicIdentifier]TopicDescriptor),
 			}
-			// Can pass in nil for execCfg to AllTargets because we're not testing
-			// database-level targets.
-			targets, err := AllTargets(context.Background(), tc.details, nil)
-			require.NoError(t, err)
 			tn, err := MakeTopicNamer(targets)
 			require.NoError(t, err)