Merge #36923

36923: changefeedccl: fix race in changeAggregator r=nvanbenschoten a=danhhz

This race was introduced in 9bde786,
which fixed a hang in `(*changeAggregator).close` which happened because
the pollerDoneCh never got closed if the RunAsyncTask rejected the work.
We could introduce some locking to fix the race, but better is to just
to close the channel if the RunAsyncTask is not going to.

Closes #35265

Release note: None

Co-authored-by: Daniel Harrison <[email protected]>

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -186,8 +186,8 @@ func (ca *changeAggregator) Start(ctx context.Context) context.Context {
 	// but only the first one is ever used.
 	ca.errCh = make(chan error, 2)
 
+	ca.pollerDoneCh = make(chan struct{})
 	if err := ca.flowCtx.Stopper().RunAsyncTask(ctx, "changefeed-poller", func(ctx context.Context) {
-		ca.pollerDoneCh = make(chan struct{})
 		defer close(ca.pollerDoneCh)
 		var err error
 		if PushEnabled.Get(&ca.flowCtx.Settings.SV) {
@@ -201,6 +201,10 @@ func (ca *changeAggregator) Start(ctx context.Context) context.Context {
 		ca.errCh <- err
 		ca.cancel()
 	}); err != nil {
+		// If err != nil then the RunAsyncTask closure never ran, which means we
+		// need to manually close ca.pollerDoneCh so `(*changeAggregator).close`
+		// doesn't hang.
+		close(ca.pollerDoneCh)
 		ca.errCh <- err
 		ca.cancel()
 	}