Merge pull request #156812 from kev-cao/blathers/backport-release-25.4-156003

release-25.4: physical: fix blocked span reconciliation jobs after stream of system

Author: kev-cao

pkg/crosscluster/physical/stream_ingestion_job_test.go

@@ -12,6 +12,7 @@ import (
 	"net"
 	"net/url"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	_ "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl"
@@ -742,3 +743,69 @@ func TestPhysicalReplicationGatewayRoute(t *testing.T) {
 	progress := jobutils.GetJobProgress(t, systemDB, jobspb.JobID(jobID))
 	require.Empty(t, progress.Details.(*jobspb.Progress_StreamIngest).StreamIngest.PartitionConnUris)
 }
+
+func TestPhysicalReplicationCancelsProducerOnCutoverFromSystem(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	// This test verifies that the span reconciliation job on a promoted standby
+	// tenant that was streaming from a system tenant is able to make progress.
+	// See https://github.com/cockroachdb/cockroach/issues/155444 for more
+	// details.
+	ctx := context.Background()
+	args := replicationtestutils.DefaultTenantStreamingClustersArgs
+	args.MultitenantSingleClusterNumNodes = 1
+	args.SrcTenantID = roachpb.SystemTenantID
+	args.SrcTenantName = "system"
+
+	c, cleanup := replicationtestutils.CreateMultiTenantStreamingCluster(ctx, t, args)
+	defer cleanup()
+
+	producerJobID, consumerJobID := c.StartStreamReplication(ctx)
+	jobutils.WaitForJobToRun(c.T, c.SrcSysSQL, jobspb.JobID(producerJobID))
+	jobutils.WaitForJobToRun(c.T, c.DestSysSQL, jobspb.JobID(consumerJobID))
+
+	replicationtestutils.WaitUntilStartTimeReached(t, c.DestSysSQL, jobspb.JobID(consumerJobID))
+	srcTime := c.SrcCluster.Server(0).Clock().Now()
+	c.Cutover(ctx, producerJobID, consumerJobID, srcTime.GoTime(), false /* async */)
+	destCleanup := c.StartDestTenant(ctx, nil /* withTestingKnobs */, 0 /* server */)
+	defer destCleanup()
+
+	// We speed up the span config reconciliation job manager so that it will
+	// quickly detect the missing reconciliation job on the tenant (the previous
+	// one was canceled as it was replicated) and spin up a new one.
+	c.DestSysSQL.Exec(t, "ALTER TENANT ALL SET CLUSTER SETTING spanconfig.reconciliation_job.check_interval = '10ms'")
+
+	registry := c.DestTenantServer.JobRegistry().(*jobs.Registry)
+	// We nudge the adoption queue to pick up the replicated stream producer job
+	// so that it will be dropped.
+	registry.TestingNudgeAdoptionQueue()
+
+	now := c.DestSysServer.Clock().Now()
+	testutils.SucceedsWithin(t, func() error {
+		var spanConfigJobID jobspb.JobID
+		rows := c.DestTenantSQL.Query(
+			t, `SELECT id FROM system.jobs WHERE job_type = 'AUTO SPAN CONFIG RECONCILIATION' AND status = 'running'`,
+		)
+		defer rows.Close()
+		if !rows.Next() {
+			return errors.New("no running span config reconciliation job found")
+		}
+		if err := rows.Scan(&spanConfigJobID); err != nil {
+			return err
+		}
+		spanConfigJob, err := registry.LoadJob(ctx, spanConfigJobID)
+		require.NoError(t, err)
+		spanConfigProg := spanConfigJob.Progress().
+			Details.(*jobspb.Progress_AutoSpanConfigReconciliation).
+			AutoSpanConfigReconciliation
+
+		if spanConfigProg.Checkpoint.Less(now) {
+			return errors.Newf(
+				"waiting for span config reconciliation job to checkpoint past %v, at %v",
+				now, spanConfigProg.Checkpoint,
+			)
+		}
+		return nil
+	}, 2*time.Minute)
+}

pkg/crosscluster/producer/producer_job.go

@@ -133,6 +133,19 @@ func (p *producerJobResumer) Resume(ctx context.Context, execCtx interface{}) er
 	jobExec := execCtx.(sql.JobExecContext)
 	execCfg := jobExec.ExecCfg()
 
+	// If the source tenant of a PCR stream is the system tenant, then the
+	// producer job will also be copied to the destination tenant, along with the
+	// associated PTS. After cutover, this blocks span config reconciliation since
+	// a tenant cannot lay a PTS on spans that another tenant owns. The producer
+	// job of course is not needed on the destination tenant, so we check for a
+	// mismatched cluster ID here and fast-exit the job.
+	if p.job.Payload().CreationClusterID != execCfg.NodeInfo.LogicalClusterID() {
+		return jobs.MarkAsPermanentJobError(errors.Newf(
+			"replication stream %d belongs to cluster %s, cannot resume on cluster %s",
+			p.job.ID(), p.job.Payload().CreationClusterID, execCfg.NodeInfo.LogicalClusterID(),
+		))
+	}
+
 	// Fire the timer immediately to start an initial progress check
 	p.timer.Reset(0)
 	for {

pkg/crosscluster/replicationtestutils/testutils.go

@@ -131,11 +131,12 @@ type TenantStreamingClusters struct {
 	SrcURL          url.URL
 	SrcCleanup      func()
 
-	DestCluster    *testcluster.TestCluster
-	DestSysServer  serverutils.ApplicationLayerInterface
-	DestSysSQL     *sqlutils.SQLRunner
-	DestTenantConn *gosql.DB
-	DestTenantSQL  *sqlutils.SQLRunner
+	DestCluster      *testcluster.TestCluster
+	DestSysServer    serverutils.ApplicationLayerInterface
+	DestSysSQL       *sqlutils.SQLRunner
+	DestTenantConn   *gosql.DB
+	DestTenantSQL    *sqlutils.SQLRunner
+	DestTenantServer serverutils.ApplicationLayerInterface
 
 	ReaderTenantSQL *sqlutils.SQLRunner
 
@@ -186,19 +187,18 @@ func (c *TenantStreamingClusters) init(ctx context.Context) {
 func (c *TenantStreamingClusters) StartDestTenant(
 	ctx context.Context, withTestingKnobs *base.TestingKnobs, server int,
 ) func() {
+	var err error
+	var testKnobs base.TestingKnobs
 	if withTestingKnobs != nil {
-		var err error
-		_, c.DestTenantConn, err = c.DestCluster.Server(server).TenantController().StartSharedProcessTenant(ctx, base.TestSharedProcessTenantArgs{
-			TenantID:    c.Args.DestTenantID,
-			TenantName:  c.Args.DestTenantName,
-			Knobs:       *withTestingKnobs,
-			UseDatabase: "defaultdb",
-		})
-		require.NoError(c.T, err)
-	} else {
-		c.DestSysSQL.Exec(c.T, `ALTER TENANT $1 START SERVICE SHARED`, c.Args.DestTenantName)
-		c.DestTenantConn = c.DestCluster.Server(server).SystemLayer().SQLConn(c.T, serverutils.DBName("cluster:"+string(c.Args.DestTenantName)+"/defaultdb"))
+		testKnobs = *withTestingKnobs
 	}
+	c.DestTenantServer, c.DestTenantConn, err = c.DestCluster.Server(server).TenantController().StartSharedProcessTenant(ctx, base.TestSharedProcessTenantArgs{
+		TenantID:    c.Args.DestTenantID,
+		TenantName:  c.Args.DestTenantName,
+		Knobs:       testKnobs,
+		UseDatabase: "defaultdb",
+	})
+	require.NoError(c.T, err)
 
 	c.DestTenantSQL = sqlutils.MakeSQLRunner(c.DestTenantConn)
 	testutils.SucceedsSoon(c.T, func() error {