changefeedccl: add retry to TestChangefeedFailOnTableOffline

Previously, this TestChangefeedFailOnTableOffline test could
fail if the import attempt failed with "result is ambiguous".
This error should be retried. This should reduce this source
of flakiness for that test.

Fixes: #142033
Release note: None

Author: aerfrei

pkg/ccl/changefeedccl/changefeed_test.go

@@ -5195,7 +5195,7 @@ func TestChangefeedFailOnTableOffline(t *testing.T) {
 		// Start an import job which will immediately pause after ingestion
 		sqlDB.Exec(t, "SET CLUSTER SETTING jobs.debug.pausepoints = 'import.after_ingest';")
 		go func() {
-			sqlDB.ExpectErrWithTimeout(t, `pause point`, `IMPORT INTO for_import CSV DATA ($1);`, dataSrv.URL)
+			sqlDB.ExpectErrWithRetry(t, `pause point`, `IMPORT INTO for_import CSV DATA ($1);`, `result is ambiguous`, dataSrv.URL)
 		}()
 		sqlDB.CheckQueryResultsRetry(
 			t,

pkg/testutils/sqlutils/BUILD.bazel

@@ -29,6 +29,7 @@ go_library(
         "//pkg/sql/sem/tree",
         "//pkg/testutils",
         "//pkg/util/protoutil",
+        "//pkg/util/retry",
         "//pkg/util/timeutil",
         "@com_github_cockroachdb_cockroach_go_v2//crdb",
         "@com_github_cockroachdb_errors//:errors",

pkg/testutils/sqlutils/sql_runner.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
+	"github.com/cockroachdb/cockroach/pkg/util/retry"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
 	"github.com/lib/pq"
@@ -244,6 +245,24 @@ func (sr *SQLRunner) ExpectErrWithTimeout(
 	}
 }
 
+// ExpectErrWithRetry wraps ExpectErr with a timeout and retry on a specific error.
+func (sr *SQLRunner) ExpectErrWithRetry(
+	t Fataler, errRE string, query string, retryableErrorRE string, args ...interface{},
+) {
+	helperOrNoop(t)()
+	var err error
+
+	retryOpts := retry.Options{MaxRetries: 5}
+	err = retryOpts.DoWithRetryable(context.Background(), func(ctx context.Context) (bool, error) {
+		_, err = sr.DB.ExecContext(context.Background(), query, args...)
+		if testutils.IsError(err, retryableErrorRE) {
+			return true, err
+		}
+		return false, err
+	})
+	sr.expectErr(t, query, err, errRE)
+}
+
 // Query is a wrapper around gosql.Query that kills the test on error.
 func (sr *SQLRunner) Query(t Fataler, query string, args ...interface{}) *gosql.Rows {
 	helperOrNoop(t)()

pkg/util/retry/retry.go

@@ -180,6 +180,29 @@ func (opts Options) Do(ctx context.Context, fn func(ctx context.Context) error)
 	return err
 }
 
+// DoWithRetryable invokes the closure according to the retry options until it
+// returns success or a non-retryable error. Always returns an error unless the
+// return is prompted by a successful invocation of `fn`.
+func (opts Options) DoWithRetryable(
+	ctx context.Context, fn func(ctx context.Context) (retryable bool, err error),
+) error {
+	var err error
+	for r := StartWithCtx(ctx, opts); r.Next(); {
+		var retryable bool
+		retryable, err = fn(ctx)
+		if err == nil {
+			return nil
+		}
+		if !retryable {
+			return err
+		}
+	}
+	if err == nil {
+		return errors.AssertionFailedf("never invoked function in DoWithRetryable")
+	}
+	return err
+}
+
 // WithMaxAttempts is a helper that runs fn N times and collects the last err.
 // The function will terminate early if the provided context is canceled, but it
 // guarantees that fn will run at least once.