changefeedccl: add test-only assertion for checkpoint fields invariant

This patch adds a test-only assertion that we won't ever see both
checkpoint fields set on a changefeed job progress struct. It is
test-only because a bug that existed on earlier versions of 25.2
could cause both checkpoint fields to be set and so the production
code continues to discard the legacy checkpoint when that happens.

Release note: None

Author: andyyang890

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -277,11 +277,15 @@ func startDistChangefeed(
 			return errors.AssertionFailedf("both legacy and current checkpoint set on " +
 				"changefeed job progress and legacy checkpoint has later timestamp")
 		}
-		// This should be an assertion failure but unfortunately due to a bug
+		// This should always be an assertion failure but unfortunately due to a bug
 		// that was included in earlier versions of 25.2 (#148620), we may fail
 		// to clear the legacy checkpoint when we start writing the new one.
 		// We instead discard the legacy checkpoint here and it will eventually be
 		// cleared once the cluster is running a newer patch release with the fix.
+		if buildutil.CrdbTestBuild {
+			return errors.AssertionFailedf("both legacy and current checkpoint set on " +
+				"changefeed job progress")
+		}
 		log.Warningf(ctx, "both legacy and current checkpoint set on changefeed job progress; "+
 			"discarding legacy checkpoint")
 		legacyCheckpoint = nil

pkg/ccl/changefeedccl/changefeed_test.go

@@ -11874,13 +11874,9 @@ func TestChangefeedResumeWithBothLegacyAndCurrentCheckpoint(t *testing.T) {
 		require.NoError(t, err)
 
 		sqlDB.Exec(t, `RESUME JOB $1`, jobFeed.JobID())
-		waitForJobState(sqlDB, t, jobFeed.JobID(), jobs.StateRunning)
-
-		// Wait for highwater to advance past the current time.
-		var tsStr string
-		sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&tsStr)
-		ts := parseTimeToHLC(t, tsStr)
-		require.NoError(t, jobFeed.WaitForHighWaterMark(ts))
+		waitForJobState(sqlDB, t, jobFeed.JobID(), jobs.StateFailed)
+		require.ErrorContains(t, jobFeed.FetchTerminalJobErr(),
+			"both legacy and current checkpoint set on changefeed job progress")
 	}
 
 	cdcTest(t, testFn, feedTestEnterpriseSinks)

pkg/cmd/roachtest/tests/mixed_version_cdc.go

@@ -672,5 +672,40 @@ func runCDCMixedVersionCheckpointing(ctx context.Context, t test.Test, c cluster
 	mvt.InMixedVersion("wait and validate", tester.waitAndValidate)
 	mvt.AfterUpgradeFinalized("wait and validate", tester.waitAndValidate)
 
-	mvt.Run()
+	// Run the test.
+	plan, err := mvt.RunE()
+	if err != nil {
+		isAffectedBy148620 := func(plan *mixedversion.TestPlan) bool {
+			if plan == nil {
+				return false
+			}
+			for _, v := range []string{"v25.2.0", "v25.2.1", "v25.2.2"} {
+				version := clusterupgrade.MustParseVersion(v)
+				for _, planVersion := range plan.Versions() {
+					if planVersion.Equal(version) {
+						return true
+					}
+				}
+			}
+			return false
+		}
+
+		isExpectedErrorDueTo148620 := func(err error) bool {
+			for _, s := range []string{
+				"both legacy and current checkpoint set on change aggregator spec",
+				"both legacy and current checkpoint set on changefeed job progress",
+			} {
+				if strings.Contains(err.Error(), s) {
+					return true
+				}
+			}
+			return false
+		}
+
+		if isAffectedBy148620(plan) && isExpectedErrorDueTo148620(err) {
+			t.Skipf("expected error due to #148620: %s", err)
+		}
+
+		t.Fatal(err)
+	}
 }