changefeedccl: Do not inhibit server shutdown.

Informs #82765

Permanet fix is being tracked by the above issue.
This is a temporary fix to ensure that change aggregators
cancel their context when the server quiesces so that the
server shutdown is not inhibited by the running changefeeds.

The test for this functionality is not being merged due
to the fact that it takes too long to run; however, the test
can be seen here: #82767

Release Notes (bug fix): Ensure running changefeeds do
not inhibit node shutdown.

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -208,17 +208,14 @@ func (ca *changeAggregator) MustBeStreaming() bool {
 
 // Start is part of the RowSource interface.
 func (ca *changeAggregator) Start(ctx context.Context) {
+	// Derive a separate context so that we can shutdown the poller.
+	ctx, ca.cancel = ca.flowCtx.Stopper().WithCancelOnQuiesce(ctx)
+
 	if ca.spec.JobID != 0 {
 		ctx = logtags.AddTag(ctx, "job", ca.spec.JobID)
 	}
 	ctx = ca.StartInternal(ctx, changeAggregatorProcName)
 
-	// Derive a separate context so that we can shutdown the poller. Note that
-	// we need to update both ctx (used throughout this function) and
-	// ProcessorBase.Ctx (used in all other methods) to the new context.
-	ctx, ca.cancel = context.WithCancel(ctx)
-	ca.Ctx = ctx
-
 	spans, err := ca.setupSpansAndFrontier()
 
 	endTime := ca.spec.Feed.EndTime

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -1030,7 +1030,12 @@ func (b *changefeedResumer) resumeWithRetries(
 			}
 		}
 
-		if !changefeedbase.IsRetryableError(err) {
+		// Retry changefeed is error is retryable.  In addition, we want to handle
+		// context cancellation as retryable, but only if the resumer context has not been cancelled.
+		// (resumer context is canceled by the jobs framework -- so we should respect it).
+		isRetryableErr := changefeedbase.IsRetryableError(err) ||
+			(ctx.Err() == nil && errors.Is(err, context.Canceled))
+		if !isRetryableErr {
 			if ctx.Err() != nil {
 				return ctx.Err()
 			}