Merge pull request #150531 from cockroachdb/blathers/backport-release-25.3-150110

msbutler · web-flow · commit 8f146ba2a286 · 2025-07-25T14:48:25.000-07:00
release-25.3: crosscluster/physical: fix status after init scan complete
diff --git a/pkg/crosscluster/physical/stream_ingestion_frontier_processor.go b/pkg/crosscluster/physical/stream_ingestion_frontier_processor.go
@@ -341,6 +341,7 @@ func (sf *streamIngestionFrontier) maybeUpdateProgress() error {
 
 		if replicatedTime.IsSet() && streamProgress.ReplicationStatus == jobspb.InitialScan {
 			streamProgress.ReplicationStatus = jobspb.Replicating
+			md.Progress.StatusMessage = streamProgress.ReplicationStatus.String()
 		}
 
 		// Keep the recorded replicatedTime empty until some advancement has been made
diff --git a/pkg/crosscluster/replicationtestutils/testutils.go b/pkg/crosscluster/replicationtestutils/testutils.go
@@ -601,6 +601,10 @@ func WaitUntilStartTimeReached(t *testing.T, db *sqlutils.SQLRunner, ingestionJo
 
 		return requireReplicatedTime(startTime, jobutils.GetJobProgress(t, db, ingestionJobID))
 	}, timeout)
+
+	var runningStatus string
+	db.QueryRow(t, "SELECT running_status FROM [SHOW JOB $1]", ingestionJobID).Scan(&runningStatus)
+	require.Equal(t, "replicating", runningStatus, "job should be in replicating state after reaching start time")
 }
 
 func WaitUntilReplicatedTime(

Original file line number	Diff line number	Diff line change
`@@ -341,6 +341,7 @@ func (sf *streamIngestionFrontier) maybeUpdateProgress() error {`
`341`	`341`
`342`	`342`	`if replicatedTime.IsSet() && streamProgress.ReplicationStatus == jobspb.InitialScan {`
`343`	`343`	`streamProgress.ReplicationStatus = jobspb.Replicating`
	`344`	`+ md.Progress.StatusMessage = streamProgress.ReplicationStatus.String()`
`344`	`345`	`}`
`345`	`346`
`346`	`347`	`// Keep the recorded replicatedTime empty until some advancement has been made`
Original file line number	Diff line number	Diff line change
`@@ -601,6 +601,10 @@ func WaitUntilStartTimeReached(t testing.T, db sqlutils.SQLRunner, ingestionJo`
`601`	`601`
`602`	`602`	`return requireReplicatedTime(startTime, jobutils.GetJobProgress(t, db, ingestionJobID))`
`603`	`603`	`}, timeout)`
	`604`	`+`
	`605`	`+ var runningStatus string`
	`606`	`+ db.QueryRow(t, "SELECT running_status FROM [SHOW JOB $1]", ingestionJobID).Scan(&runningStatus)`
	`607`	`+ require.Equal(t, "replicating", runningStatus, "job should be in replicating state after reaching start time")`
`604`	`608`	`}`
`605`	`609`
`606`	`610`	`func WaitUntilReplicatedTime(`