changefeedccl: add on_error option to pause changefeeds on failure

[spiffyy99]

Previously, changefeeds always failed when encountering a non-
retryable error. This option allows the user to pause on failure
and resume later, while still failing as default behavior.

Release note (enterprise change): new 'on_error' option to pause
on non-retryable errors instead of failing.

[cockroach-teamcity]

This change is 

[spiffyy99]

outdated comment: we can hit this if the retry count is exhausted as well

[miretskiy]

Very nice!
Just few minor nits/questions.

Reviewed 6 of 6 files at r1.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @spiffyyeng)

---

pkg/ccl/changefeedccl/changefeed_stmt.go, line 555 at r1 (raw file):

		default:
			return jobspb.ChangefeedDetails{}, errors.Errorf(
				`unknown %s: %s`, opt, v)

I would add maybe a a hint (errors.WithHint), or just expand the error message with "valid values are....."

---

pkg/ccl/changefeedccl/changefeed_stmt.go, line 629 at r1 (raw file):

				return pauseErr
			}
			// TODO (ryan min): Populate pause reason with error once column is added (#67928)

We already have that. See running status (see changefeed_processors).
In order to update run status and indicate that we're pausing because of this policy, as opposed to user initiated action, we will probably need to export pauseRequested method (as well as onPauseRequestFunc) so that we can directly call b.job.PauseRequested, and update run status in the passed in function.

---

pkg/ccl/changefeedccl/changefeed_stmt.go, line 633 at r1 (raw file):

Quoted 18 lines of code…

		switch onError := changefeedbase.OnErrorType(details.Opts[changefeedbase.OptOnError]); onError {
		// default behavior
		case changefeedbase.OptOnErrorFail:
			return err
		// pause instead of failing
		case changefeedbase.OptOnErrorPause:
			// note: we only want the job to pause here if a failure happens, not a
			// user-initiated cancellation. if the job has been canceled, the ctx
			// will handle it and the pause will return an error.
			pauseErr := execCfg.JobRegistry.PauseRequested(ctx, jobExec.ExtendedEvalContext().Txn, jobID)
			if pauseErr != nil {
				return pauseErr
			}
			// TODO (ryan min): Populate pause reason with error once column is added (#67928)
		default:
			return errors.Errorf("unrecognized option value: %s=%s",
				changefeedbase.OptOnError, details.Opts[changefeedbase.OptOnError])
		}

I would move this into a helper handleChangefeedError() which returns an error (or nil)..
So, this code becomes:

if err != nil {
   err = handleChangefeedError(....)
}
return err

---

pkg/ccl/changefeedccl/changefeed_test.go, line 4265 at r1 (raw file):

			// check for paused status on failure
			err := feedJob.WaitForStatus(func(s jobs.Status) bool { return s == jobs.StatusPaused })

require.NoError(t, feedJob.WaitForStatus...)?

---

pkg/ccl/changefeedccl/changefeed_test.go, line 4345 at r1 (raw file):

	t.Run(`cloudstorage`, cloudStorageTest(testFn))
	t.Run(`kafka`, kafkaTest(testFn))
	t.Run(`webhook`, webhookTest(testFn))

nice test.

[spiffyy99]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @miretskiy and @spiffyyeng)

---

pkg/ccl/changefeedccl/changefeed_stmt.go, line 555 at r1 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

I would add maybe a a hint (errors.WithHint), or just expand the error message with "valid values are....."

Done.

---

pkg/ccl/changefeedccl/changefeed_stmt.go, line 629 at r1 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

We already have that. See running status (see changefeed_processors).
In order to update run status and indicate that we're pausing because of this policy, as opposed to user initiated action, we will probably need to export pauseRequested method (as well as onPauseRequestFunc) so that we can directly call b.job.PauseRequested, and update run status in the passed in function.

discussed offline. run status is updated internally in the progress column but is not visible in SHOW JOBS when paused.

---

pkg/ccl/changefeedccl/changefeed_stmt.go, line 633 at r1 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

		switch onError := changefeedbase.OnErrorType(details.Opts[changefeedbase.OptOnError]); onError {
		// default behavior
		case changefeedbase.OptOnErrorFail:
			return err
		// pause instead of failing
		case changefeedbase.OptOnErrorPause:
			// note: we only want the job to pause here if a failure happens, not a
			// user-initiated cancellation. if the job has been canceled, the ctx
			// will handle it and the pause will return an error.
			pauseErr := execCfg.JobRegistry.PauseRequested(ctx, jobExec.ExtendedEvalContext().Txn, jobID)
			if pauseErr != nil {
				return pauseErr
			}
			// TODO (ryan min): Populate pause reason with error once column is added (#67928)
		default:
			return errors.Errorf("unrecognized option value: %s=%s",
				changefeedbase.OptOnError, details.Opts[changefeedbase.OptOnError])
		}

I would move this into a helper handleChangefeedError() which returns an error (or nil)..
So, this code becomes:

if err != nil {
   err = handleChangefeedError(....)
}
return err

Done.

---

pkg/ccl/changefeedccl/changefeed_test.go, line 4265 at r1 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

require.NoError(t, feedJob.WaitForStatus...)?

Done.

[miretskiy]

Reviewed 1 of 4 files at r3.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @miretskiy and @spiffyyeng)

---

pkg/ccl/changefeedccl/changefeed_stmt.go, line 646 at r3 (raw file):

			// directly update running status to avoid the running/reverted job status check
			progress.RunningStatus = fmt.Sprintf("paused on error: %v", changefeedErr)
			log.Warningf(ctx, progress.RunningStatus)

nit: Let's be more verbose here

log.Warningf(ctx, "job failed (%v) but is being paused because of on_error=pause", err)

[spiffyy99]

Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @miretskiy and @spiffyyeng)

---

pkg/ccl/changefeedccl/changefeed_stmt.go, line 646 at r3 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

nit: Let's be more verbose here

log.Warningf(ctx, "job failed (%v) but is being paused because of on_error=pause", err)

Done.

[spiffyy99]

TFTR!

[spiffyy99]

bors r+

[craig]

Build succeeded:

• GitHub CI (Cockroach)