changefeedccl: fix db-level feed error handling in changefeed processors

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

Author: KeithCh

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)