changefeedccl: add cluster setting for changefeed max retry backoff

When changefeeds enter a high-level retry loop, e.g. as part of a
rolling restart, there is an exponential backoff applied. The default
max backoff was 10m, but due to considerations in #146448 a lower 1m
max was considered for some rolling restart cases. This PR makes the max
backoff configurable via the non-pubic cluster setting
changefeed.max_retry_backoff, so that most users can keep the old
setting of 10m, which is better for degenerate scenarios when
changefeeds might try to retry frequently due to cluster instability.
It also adds a separate cluster setting, changefeed.retry_backoff_reset,
which is the amount of time between retries before the backoff timer
resets. Both settings have a default of 10m.

Epic: none
Fixes: #148467

Release note: None

Author: rharding6373

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -411,7 +411,9 @@ func coreChangefeed(
 	p.ExtendedEvalContext().ChangefeedState = localState
 	knobs, _ := p.ExecCfg().DistSQLSrv.TestingKnobs.Changefeed.(*TestingKnobs)
 
-	for r := getRetry(ctx); ; {
+	maxBackoff := changefeedbase.MaxRetryBackoff.Get(&p.ExecCfg().Settings.SV)
+	backoffReset := changefeedbase.RetryBackoffReset.Get(&p.ExecCfg().Settings.SV)
+	for r := getRetry(ctx, maxBackoff, backoffReset); ; {
 		if !r.Next() {
 			// Retry loop exits when context is canceled.
 			log.Infof(ctx, "core changefeed retry loop exiting: %s", ctx.Err())
@@ -1441,7 +1443,9 @@ func (b *changefeedResumer) resumeWithRetries(
 		b.mu.perNodeAggregatorStats[componentID] = *meta
 	}
 
-	for r := getRetry(ctx); r.Next(); {
+	maxBackoff := changefeedbase.MaxRetryBackoff.Get(&execCfg.Settings.SV)
+	backoffReset := changefeedbase.RetryBackoffReset.Get(&execCfg.Settings.SV)
+	for r := getRetry(ctx, maxBackoff, backoffReset); r.Next(); {
 		flowErr := maybeUpgradePreProductionReadyExpression(ctx, jobID, details, jobExec)
 
 		if flowErr == nil {

pkg/ccl/changefeedccl/changefeedbase/settings.go

@@ -349,3 +349,23 @@ var Quantize = settings.RegisterDurationSettingWithExplicitUnit(
 	time.Duration(metamorphic.ConstantWithTestRange("changefeed.resolved_timestamp.granularity", 1, 0, 10))*time.Second,
 	settings.DurationWithMinimum(0),
 )
+
+// MaxRetryBackoff is the maximum time a changefeed will backoff when in
+// a top-level retry loop, for example during rolling restarts.
+var MaxRetryBackoff = settings.RegisterDurationSettingWithExplicitUnit(
+	settings.ApplicationLevel,
+	"changefeed.max_retry_backoff",
+	"the maximum time a changefeed will backoff when retrying after a restart and how long between retries before backoff resets",
+	10*time.Minute, /* defaultValue */
+	settings.DurationInRange(1*time.Second, 1*time.Hour),
+)
+
+// RetryBackoffReset is the time between changefeed retries before the
+// backoff timer resets.
+var RetryBackoffReset = settings.RegisterDurationSettingWithExplicitUnit(
+	settings.ApplicationLevel,
+	"changefeed.retry_backoff_reset",
+	"the time between changefeed retries before the backoff timer resets",
+	10*time.Minute, /* defaultValue */
+	settings.DurationInRange(1*time.Second, 1*time.Hour),
+)

pkg/ccl/changefeedccl/retry.go

@@ -18,11 +18,11 @@ var useFastRetry = envutil.EnvOrDefaultBool(
 	"COCKROACH_CHANGEFEED_TESTING_FAST_RETRY", false)
 
 // getRetry returns retry object for changefeed.
-func getRetry(ctx context.Context) Retry {
+func getRetry(ctx context.Context, maxBackoff, backoffReset time.Duration) Retry {
 	opts := retry.Options{
 		InitialBackoff: 1 * time.Second,
 		Multiplier:     2,
-		MaxBackoff:     1 * time.Minute,
+		MaxBackoff:     maxBackoff,
 	}
 
 	if useFastRetry {
@@ -33,7 +33,8 @@ func getRetry(ctx context.Context) Retry {
 		}
 	}
 
-	return Retry{Retry: retry.StartWithCtx(ctx, opts)}
+	return Retry{Retry: retry.StartWithCtx(ctx, opts),
+		resetRetryAfter: backoffReset}
 }
 
 func testingUseFastRetry() func() {
@@ -43,16 +44,15 @@ func testingUseFastRetry() func() {
 	}
 }
 
-// reset retry state after changefeed ran for that much time
-// without errors.
-const resetRetryAfter = 10 * time.Minute
-
 // Retry is a thin wrapper around retry.Retry which
 // resets retry state if changefeed been running for sufficiently
 // long time.
 type Retry struct {
 	retry.Retry
 	lastRetry time.Time
+	// reset retry state after changefeed ran for that much time
+	// without errors.
+	resetRetryAfter time.Duration
 }
 
 // Next returns whether the retry loop should continue, and blocks for the
@@ -63,7 +63,7 @@ func (r *Retry) Next() bool {
 	defer func() {
 		r.lastRetry = timeutil.Now()
 	}()
-	if timeutil.Since(r.lastRetry) > resetRetryAfter {
+	if timeutil.Since(r.lastRetry) > r.resetRetryAfter {
 		r.Reset()
 	}
 	return r.Retry.Next()