Merge cockroachdb#69933

69933: changefeedccl: make scanrequestlimit dynamic r=samiskin a=samiskin

Resolves cockroachdb#68966

Figured there's enough work still to get to that I won't add a test to this minor change for now.

Release justification: very low-risk minor ux change

Previously, the ScanRequestLimit cluster setting that controlled the
max number of concurrent backfill workers per node was only read once at
the beginning of a backfill job.

If a user wanted a backfill to proceed more quickly / slowly, they'd
have to restart the entire backfill job.

This change just makes the backfill workers read the current
scanrequestlimit every time a new worker is about to begin a scan.

Release note (bug fix): Backfills will now always respect the most
up-to-date value of "changefeed.backfill.concurrent_scan_requests" even
during an ongoing backfill.

Co-authored-by: Shiranka Miskin <[email protected]>

pkg/ccl/changefeedccl/kvfeed/scanner.go

@@ -67,11 +67,21 @@ func (p *scanRequestScanner) Scan(
 
 	maxConcurrentExports := maxConcurrentExportRequests(p.gossip, &p.settings.SV)
 	exportLim := limit.MakeConcurrentRequestLimiter("changefeedExportRequestLimiter", maxConcurrentExports)
+
+	lastScanLimitUserSetting := changefeedbase.ScanRequestLimit.Get(&p.settings.SV)
+
 	g := ctxgroup.WithContext(ctx)
 	// atomicFinished is used only to enhance debugging messages.
 	var atomicFinished int64
 	for _, span := range spans {
 		span := span
+
+		// If the user defined scan request limit has changed, recalculate it
+		if currentUserScanLimit := changefeedbase.ScanRequestLimit.Get(&p.settings.SV); currentUserScanLimit != lastScanLimitUserSetting {
+			lastScanLimitUserSetting = currentUserScanLimit
+			exportLim.SetLimit(maxConcurrentExportRequests(p.gossip, &p.settings.SV))
+		}
+
 		limAlloc, err := exportLim.Begin(ctx)
 		if err != nil {
 			cancel()