Merge #95370 #96014 #96111 #96125 #96206 #96499 #96681 #96689

95370: logictest: add a skip_on_retry directive to the sequences test r=ajwerner a=ajwerner

See [this bors run](https://teamcity.cockroachdb.com/buildConfiguration/Cockroach_BazelEssentialCi/8344646?showRootCauses=false&expandBuildChangesSection=true&expandBuildProblemsSection=true&expandBuildTestsSection=true)

In the long run, this directive is not great, but it's better than flakes.

Epic: none

Release note: None

96014: multitenant: prevent tenant from overriding next tenant's span config r=arulajmani a=ecwall

Fixes #95882

We want to ensure we have a split at the non-inclusive end of the tenant's
keyspace, which also happens to be the inclusive start of the next
tenant's (with ID=ours+1). If we didn't do anything here, and we were the
tenant with the highest ID thus far, our last range would extend to /Max.
If a tenant with a higher ID was created, when installing its initial span
config record, it would carve itself off (and our last range would only
extend to that next tenant's boundary), but this is a bit awkward for code
that wants to rely on the invariant that ranges for a tenant only extend
to the tenant's well-defined end key.

So how do we ensure this split at this tenant's non-inclusive end key?
Hard splits are controlled by the start keys on span config records[^1],
so we'll try insert one accordingly. But we cannot do this blindly. We
cannot assume that tenants are created in ID order, so it's possible that
the tenant with the next ID is already present + running. If so, it may
already have span config records that start at the key at which we want
to write a span config record for. Over-writing it blindly would be a
mistake -- there's no telling what config that next tenant has associated
for that span. So we'll do something simple -- we'll check transactionally
whether there's anything written already, and if so, do nothing. We
already have the split we need.

[^1]: See ComputeSplitKey in spanconfig.StoreReader.

Release note: None

96111: logictest: extend temp_table test to exercise discard + drop r=ajwerner a=ajwerner

We had bugs with this in earlier releases. This is a forward-port of a test introduced in #96102.

It never failed master, but no reason to ever let this regress.

Epic: none

Release note: None

96125: logictest: frontport test from #96124 r=ajwerner a=ajwerner

This testing is worth having. It does pass.

Epic: None

Release note: None

96206: streamingccl,db-console: add Cross-Cluster Replication metrics page r=adityamaru a=adityamaru

This change adds a new option to the metric dropdown dashboard
for Cross-Cluster Replication. This page will show all the user-facing
metrics relevant to C2C replication.

Informs: #95995

Release note: None

server,db-console: add feature flag for Cross-Cluster Replication dashboard

This change adds a feature flag whose value is determined by the cluster
setting `cross_cluster_replication.enabled`. This feature flag is then used
when rendering the metrics dashboard dropdown options to decide whether to
display the cross-cluster replication dashboard.

This change also deletes the session variable and associated cluster setting
`sql.defaults.experimental_stream_replication.enabled` so as to unify all
interactions with cross-cluster replication under a single cluster setting.

Fixes: #95995

Release note: None

96499: sql: deflake TestParseClientProvidedSessionParameters r=knz a=rafiss

fixes #96492

pgx starts an internal goroutine if a context is cancelled. We need to wait for it to be done in order to avoid a leaked goroutine in the test.

Release note: None

96681: *: Properly support partial UNIQUE WITHOUT INDEX referencing type descs r=Xiang-Gu a=Xiang-Gu

Previously if a partial UNIQUE WITHOUT INDEX references a type descriptor in its predicate, we didn't add back-references in the type descriptor, both in the legacy and declarative schema changer. This commit fixes this.

Fixes #96678
Release note: None


96689: roachtest: return error in StopCockroachGracefullyOnNode r=herkolategan,srosenberg a=renatolabs

That function misleadingly returned an (always nil) error, calling `t.Fatal()` functions in it. The calls to `Stop` have been replaced with calls to `StopE`.

Epic: none

Release note: None

Co-authored-by: Andrew Werner <[email protected]>
Co-authored-by: Evan Wall <[email protected]>
Co-authored-by: Andrew Werner <[email protected]>
Co-authored-by: adityamaru <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>
Co-authored-by: Xiang Gu <[email protected]>
Co-authored-by: Renato Costa <[email protected]>

docs/generated/settings/settings-for-tenants.txt

@@ -134,9 +134,6 @@ This session variable default should now be configured using ALTER ROLE... SET:
 sql.defaults.experimental_implicit_column_partitioning.enabled	boolean	false	"default value for experimental_enable_temp_tables; allows for the use of implicit column partitioning
 This cluster setting is being kept to preserve backwards-compatibility.
 This session variable default should now be configured using ALTER ROLE... SET: https://www.cockroachlabs.com/docs/stable/alter-role.html"
-sql.defaults.experimental_stream_replication.enabled	boolean	false	"default value for experimental_stream_replication session setting;enables the ability to setup a replication stream
-This cluster setting is being kept to preserve backwards-compatibility.
-This session variable default should now be configured using ALTER ROLE... SET: https://www.cockroachlabs.com/docs/stable/alter-role.html"
 sql.defaults.experimental_temporary_tables.enabled	boolean	false	"default value for experimental_enable_temp_tables; allows for use of temporary tables by default
 This cluster setting is being kept to preserve backwards-compatibility.
 This session variable default should now be configured using ALTER ROLE... SET: https://www.cockroachlabs.com/docs/stable/alter-role.html"

docs/generated/settings/settings.html

@@ -135,7 +135,6 @@
 <tr><td><div id="setting-sql-defaults-experimental-distsql-planning" class="anchored"><code>sql.defaults.experimental_distsql_planning</code></div></td><td>enumeration</td><td><code>off</code></td><td>default experimental_distsql_planning mode; enables experimental opt-driven DistSQL planning [off = 0, on = 1]<br/>This cluster setting is being kept to preserve backwards-compatibility.<br/>This session variable default should now be configured using <a href="alter-role.html"><code>ALTER ROLE... SET</code></a></td></tr>
 <tr><td><div id="setting-sql-defaults-experimental-enable-unique-without-index-constraints-enabled" class="anchored"><code>sql.defaults.experimental_enable_unique_without_index_constraints.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>default value for experimental_enable_unique_without_index_constraints session setting;disables unique without index constraints by default<br/>This cluster setting is being kept to preserve backwards-compatibility.<br/>This session variable default should now be configured using <a href="alter-role.html"><code>ALTER ROLE... SET</code></a></td></tr>
 <tr><td><div id="setting-sql-defaults-experimental-implicit-column-partitioning-enabled" class="anchored"><code>sql.defaults.experimental_implicit_column_partitioning.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>default value for experimental_enable_temp_tables; allows for the use of implicit column partitioning<br/>This cluster setting is being kept to preserve backwards-compatibility.<br/>This session variable default should now be configured using <a href="alter-role.html"><code>ALTER ROLE... SET</code></a></td></tr>
-<tr><td><div id="setting-sql-defaults-experimental-stream-replication-enabled" class="anchored"><code>sql.defaults.experimental_stream_replication.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>default value for experimental_stream_replication session setting;enables the ability to setup a replication stream<br/>This cluster setting is being kept to preserve backwards-compatibility.<br/>This session variable default should now be configured using <a href="alter-role.html"><code>ALTER ROLE... SET</code></a></td></tr>
 <tr><td><div id="setting-sql-defaults-experimental-temporary-tables-enabled" class="anchored"><code>sql.defaults.experimental_temporary_tables.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>default value for experimental_enable_temp_tables; allows for use of temporary tables by default<br/>This cluster setting is being kept to preserve backwards-compatibility.<br/>This session variable default should now be configured using <a href="alter-role.html"><code>ALTER ROLE... SET</code></a></td></tr>
 <tr><td><div id="setting-sql-defaults-foreign-key-cascades-limit" class="anchored"><code>sql.defaults.foreign_key_cascades_limit</code></div></td><td>integer</td><td><code>10000</code></td><td>default value for foreign_key_cascades_limit session setting; limits the number of cascading operations that run as part of a single query<br/>This cluster setting is being kept to preserve backwards-compatibility.<br/>This session variable default should now be configured using <a href="alter-role.html"><code>ALTER ROLE... SET</code></a></td></tr>
 <tr><td><div id="setting-sql-defaults-idle-in-session-timeout" class="anchored"><code>sql.defaults.idle_in_session_timeout</code></div></td><td>duration</td><td><code>0s</code></td><td>default value for the idle_in_session_timeout; default value for the idle_in_session_timeout session setting; controls the duration a session is permitted to idle before the session is terminated; if set to 0, there is no timeout<br/>This cluster setting is being kept to preserve backwards-compatibility.<br/>This session variable default should now be configured using <a href="alter-role.html"><code>ALTER ROLE... SET</code></a></td></tr>

pkg/ccl/spanconfigccl/spanconfigreconcilerccl/testdata/multitenant/tenant_end_key_split

@@ -0,0 +1,115 @@
+# Test that tenant creation does not overwrite span config state for another
+# tenant.
+
+reconcile
+----
+
+mutations discard
+----
+
+# Initialize a new tenant, tenant=11, that DOES NOT have a pre-existing tenant,
+# tenant=12, next to it.
+initialize tenant=11
+----
+
+# A record IS written for a key that logically belongs to the next tenant,
+# tenant=12, because tenant=12 DOES NOT exist.
+state offset=57 limit=3
+----
+...
+/Tenant/11{-\x00}                          database system (tenant)
+/Tenant/12{-\x00}                          database system (tenant)
+
+# Start the reconciliation loop for the tenant=11. It'll first clear out its own
+# first-key record and install span configs for its SQL state. It won't touch
+# the record we created at in the tenant=12 keyspace because it's not taught to
+# look there, and neither should it. The keyspace it's responsible for is its
+# own.
+reconcile tenant=11
+----
+
+# Peek near the start of the span_configurations table where tenant=11's records
+# are stored. The first one is from the start of its keyspace to start of
+# table with ID=4: /Tenant/11{-/Table/4}.
+state offset=56 limit=3
+----
+...
+/Table/5{7-8}                              ttl_seconds=3600 ignore_strict_gc=true num_replicas=5 rangefeed_enabled=true
+/Tenant/11{-/Table/4}                      database system (tenant)
+/Tenant/11/Table/{4-5}                     database system (tenant)
+...
+
+# Peek near the end of the span_configurations table where tenant=11's records
+# are stored. The last one is for its last system table:
+# /Tenant/11/Table/5{7-8}. Right now the split is at /Tenant/12. Which is fine.
+state offset=99 limit=3
+----
+...
+/Tenant/11/Table/5{7-8}                    database system (tenant)
+/Tenant/12{-\x00}                          database system (tenant)
+
+# Just another view of what the tenant's reconciler actually did. It got rid of
+# the original, single-key /Tenant/11{-\x00} record, and replaced it with
+# /Tenant/11{-/Table/4}, just like the comment above said. The remaining upserts
+# are for its SQL state.
+mutations tenant=11 limit=2
+----
+delete /Tenant/11{-\x00}
+upsert /Tenant/11{-/Table/4}               database system (tenant)
+upsert /Tenant/11/Table/{4-5}              database system (tenant)
+upsert /Tenant/11/Table/{5-6}              database system (tenant)
+upsert /Tenant/11/Table/{6-7}              database system (tenant)
+upsert /Tenant/11/Table/{7-8}              database system (tenant)
+upsert /Tenant/11/Table/1{1-2}             database system (tenant)
+upsert /Tenant/11/Table/1{2-3}             database system (tenant)
+upsert /Tenant/11/Table/1{3-4}             database system (tenant)
+upsert /Tenant/11/Table/1{4-5}             database system (tenant)
+upsert /Tenant/11/Table/1{5-6}             database system (tenant)
+upsert /Tenant/11/Table/{19-20}            database system (tenant)
+upsert /Tenant/11/Table/2{0-1}             database system (tenant)
+upsert /Tenant/11/Table/2{1-2}             database system (tenant)
+upsert /Tenant/11/Table/2{3-4}             database system (tenant)
+upsert /Tenant/11/Table/2{4-5}             database system (tenant)
+upsert /Tenant/11/Table/2{5-6}             database system (tenant)
+upsert /Tenant/11/Table/2{6-7}             database system (tenant)
+upsert /Tenant/11/Table/2{7-8}             database system (tenant)
+upsert /Tenant/11/Table/2{8-9}             database system (tenant)
+upsert /Tenant/11/NamespaceTable/{30-Max}  database system (tenant)
+upsert /Tenant/11/{NamespaceTable/Max-Table/32} database system (tenant)
+upsert /Tenant/11/Table/3{2-3}             database system (tenant)
+upsert /Tenant/11/Table/3{3-4}             database system (tenant)
+upsert /Tenant/11/Table/3{4-5}             database system (tenant)
+upsert /Tenant/11/Table/3{5-6}             database system (tenant)
+upsert /Tenant/11/Table/3{6-7}             database system (tenant)
+upsert /Tenant/11/Table/3{7-8}             database system (tenant)
+upsert /Tenant/11/Table/{39-40}            database system (tenant)
+upsert /Tenant/11/Table/4{0-1}             database system (tenant)
+upsert /Tenant/11/Table/4{1-2}             database system (tenant)
+upsert /Tenant/11/Table/4{2-3}             database system (tenant)
+upsert /Tenant/11/Table/4{3-4}             database system (tenant)
+upsert /Tenant/11/Table/4{4-5}             database system (tenant)
+upsert /Tenant/11/Table/4{6-7}             database system (tenant)
+upsert /Tenant/11/Table/4{8-9}             database system (tenant)
+upsert /Tenant/11/Table/5{0-1}             database system (tenant)
+upsert /Tenant/11/Table/5{1-2}             database system (tenant)
+upsert /Tenant/11/Table/5{2-3}             database system (tenant)
+upsert /Tenant/11/Table/5{3-4}             database system (tenant)
+upsert /Tenant/11/Table/5{4-5}             database system (tenant)
+upsert /Tenant/11/Table/5{5-6}             database system (tenant)
+upsert /Tenant/11/Table/5{6-7}             database system (tenant)
+upsert /Tenant/11/Table/5{7-8}             database system (tenant)
+
+# Initialize a new tenant, tenant=10, that DOES have a pre-existing tenant,
+# tenant=11, next to it.
+initialize tenant=10
+----
+
+# A record IS NOT written for a key that logically belongs to the next tenant,
+# tenant=11, because tenant=11 DOES exist.
+state offset=57 limit=3
+----
+...
+/Tenant/10{-\x00}                          database system (tenant)
+/Tenant/11{-/Table/4}                      database system (tenant)
+/Tenant/11/Table/{4-5}                     database system (tenant)
+...

pkg/ccl/streamingccl/replicationtestutils/replication_helpers.go

@@ -223,7 +223,7 @@ func NewReplicationHelper(
 SET CLUSTER SETTING kv.rangefeed.enabled = true;
 SET CLUSTER SETTING kv.closed_timestamp.target_duration = '1s';
 SET CLUSTER SETTING changefeed.experimental_poll_interval = '10ms';
-SET CLUSTER SETTING sql.defaults.experimental_stream_replication.enabled = 'on';
+SET CLUSTER SETTING cross_cluster_replication.enabled = true;
 `, `;`)...)
 
 	// Sink to read data from.

pkg/ccl/streamingccl/replicationtestutils/testutils.go

@@ -273,7 +273,7 @@ func CreateTenantStreamingClusters(
 		args.DestInitFunc(t, tsc.DestSysSQL)
 	}
 	// Enable stream replication on dest by default.
-	tsc.DestSysSQL.Exec(t, `SET enable_experimental_stream_replication = true;`)
+	tsc.DestSysSQL.Exec(t, `SET CLUSTER SETTING cross_cluster_replication.enabled = true;`)
 	return tsc, func() {
 		require.NoError(t, srcTenantConn.Close())
 		destCleanup()

pkg/ccl/streamingccl/settings.go

@@ -14,6 +14,15 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/settings"
 )
 
+// CrossClusterReplicationEnabled enables the ability to setup and control a
+// cross cluster replication stream.
+var CrossClusterReplicationEnabled = settings.RegisterBoolSetting(
+	settings.TenantWritable,
+	"cross_cluster_replication.enabled",
+	"enables the ability to setup and control a cross cluster replication stream",
+	false,
+)
+
 // StreamReplicationStreamLivenessTrackFrequency controls frequency to check
 // the liveness of a streaming replication producer job.
 var StreamReplicationStreamLivenessTrackFrequency = settings.RegisterDurationSetting(

pkg/ccl/streamingccl/streamingest/alter_replication_job.go

@@ -12,6 +12,7 @@ import (
 	"context"
 	"math"
 
+	"github.com/cockroachdb/cockroach/pkg/ccl/streamingccl"
 	"github.com/cockroachdb/cockroach/pkg/ccl/streamingccl/replicationutils"
 	"github.com/cockroachdb/cockroach/pkg/ccl/utilccl"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
@@ -111,6 +112,19 @@ func alterReplicationJobHook(
 		return nil, nil, nil, false, nil
 	}
 
+	if !streamingccl.CrossClusterReplicationEnabled.Get(&p.ExecCfg().Settings.SV) {
+		return nil, nil, nil, false, errors.WithTelemetry(
+			pgerror.WithCandidateCode(
+				errors.WithHint(
+					errors.Newf("cross cluster replication is disabled"),
+					"You can enable cross cluster replication by running `SET CLUSTER SETTING cross_cluster_replication.enabled = true`.",
+				),
+				pgcode.ExperimentalFeature,
+			),
+			"cross_cluster_replication.enabled",
+		)
+	}
+
 	if err := p.RequireAdminRole(ctx, "ALTER TENANT REPLICATION"); err != nil {
 		return nil, nil, nil, false, err
 	}

pkg/ccl/streamingccl/streamingest/alter_replication_job_test.go

@@ -85,6 +85,7 @@ func TestAlterTenantPauseResume(t *testing.T) {
 	})
 
 	t.Run("pause-resume-as-non-system-tenant", func(t *testing.T) {
+		c.DestTenantSQL.Exec(t, `SET CLUSTER SETTING cross_cluster_replication.enabled = true`)
 		c.DestTenantSQL.ExpectErr(t, "only the system tenant can alter tenant", `ALTER TENANT $1 PAUSE REPLICATION`, "foo")
 		c.DestTenantSQL.ExpectErr(t, "only the system tenant can alter tenant", `ALTER TENANT $1 RESUME REPLICATION`, "foo")
 	})

pkg/ccl/streamingccl/streamingest/metrics.go

@@ -24,31 +24,31 @@ const (
 var (
 	metaReplicationEventsIngested = metric.Metadata{
 		Name:        "replication.events_ingested",
-		Help:        "Events ingested by all ingestion jobs",
+		Help:        "Events ingested by all replication jobs",
 		Measurement: "Events",
 		Unit:        metric.Unit_COUNT,
 	}
 	metaReplicationResolvedEventsIngested = metric.Metadata{
 		Name:        "replication.resolved_events_ingested",
-		Help:        "Resolved events ingested by all ingestion jobs",
+		Help:        "Resolved events ingested by all replication jobs",
 		Measurement: "Events",
 		Unit:        metric.Unit_COUNT,
 	}
 	metaReplicationIngestedBytes = metric.Metadata{
-		Name:        "replication.ingested_bytes",
-		Help:        "Bytes ingested by all ingestion jobs",
+		Name:        "replication.logical_bytes",
+		Help:        "Logical bytes (sum of keys + values) ingested by all replication jobs",
 		Measurement: "Bytes",
 		Unit:        metric.Unit_BYTES,
 	}
 	metaReplicationSSTBytes = metric.Metadata{
 		Name:        "replication.sst_bytes",
-		Help:        "SST bytes (compressed) sent to KV by all ingestion jobs",
+		Help:        "SST bytes (compressed) sent to KV by all replication jobs",
 		Measurement: "Bytes",
 		Unit:        metric.Unit_BYTES,
 	}
 	metaReplicationFlushes = metric.Metadata{
 		Name:        "replication.flushes",
-		Help:        "Total flushes across all ingestion jobs",
+		Help:        "Total flushes across all replication jobs",
 		Measurement: "Flushes",
 		Unit:        metric.Unit_COUNT,
 	}
@@ -105,8 +105,9 @@ var (
 		Unit:        metric.Unit_COUNT,
 	}
 	metaFrontierLagSeconds = metric.Metadata{
-		Name:        "replication.frontier_lag_seconds",
-		Help:        "Time the replication frontier lags",
+		Name: "replication.frontier_lag_seconds",
+		Help: "Time between the wall clock and replicated time of the replication stream. " +
+			"This metric tracks how far behind the replication stream is relative to now",
 		Measurement: "Seconds",
 		Unit:        metric.Unit_SECONDS,
 	}

pkg/ccl/streamingccl/streamingest/replication_random_client_test.go

@@ -238,9 +238,9 @@ func TestStreamIngestionJobWithRandomClient(t *testing.T) {
 
 	// Attempt to run the ingestion job without enabling the experimental setting.
 	_, err = conn.Exec(query)
-	require.True(t, testutils.IsError(err, "stream replication is only supported experimentally"))
+	require.True(t, testutils.IsError(err, "cross cluster replication is disabled"))
 
-	_, err = conn.Exec(`SET enable_experimental_stream_replication = true`)
+	_, err = conn.Exec(`SET CLUSTER SETTING cross_cluster_replication.enabled = true;`)
 	require.NoError(t, err)
 
 	_, err = conn.Exec(query)

pkg/ccl/streamingccl/streamingest/stream_ingestion_job_test.go

@@ -113,7 +113,7 @@ SET CLUSTER SETTING stream_replication.consumer_heartbeat_frequency = '100ms';
 SET CLUSTER SETTING bulkio.stream_ingestion.minimum_flush_interval = '500ms';
 SET CLUSTER SETTING bulkio.stream_ingestion.cutover_signal_poll_interval = '100ms';
 SET CLUSTER SETTING stream_replication.job_checkpoint_frequency = '100ms';
-SET enable_experimental_stream_replication = true;
+SET CLUSTER SETTING cross_cluster_replication.enabled = true;
 `,
 		";")...)
 
@@ -191,7 +191,7 @@ func TestTenantStreamingCreationErrors(t *testing.T) {
 	SrcSysSQL.Exec(t, `SET CLUSTER SETTING kv.rangefeed.enabled = true`)
 
 	DestSysSQL := sqlutils.MakeSQLRunner(destDB)
-	DestSysSQL.Exec(t, `SET enable_experimental_stream_replication = true`)
+	DestSysSQL.Exec(t, `SET CLUSTER SETTING cross_cluster_replication.enabled = true;`)
 
 	// Sink to read data from.
 	srcPgURL, cleanupSink := sqlutils.PGUrl(t, srcServer.ServingSQLAddr(), t.Name(), url.User(username.RootUser))

pkg/ccl/streamingccl/streamingest/stream_ingestion_planning.go

@@ -68,17 +68,16 @@ func ingestionPlanHook(
 		return nil, nil, nil, false, nil
 	}
 
-	// Check if the experimental feature is enabled.
-	if !p.SessionData().EnableStreamReplication {
+	if !streamingccl.CrossClusterReplicationEnabled.Get(&p.ExecCfg().Settings.SV) {
 		return nil, nil, nil, false, errors.WithTelemetry(
 			pgerror.WithCandidateCode(
 				errors.WithHint(
-					errors.Newf("stream replication is only supported experimentally"),
-					"You can enable stream replication by running `SET enable_experimental_stream_replication = true`.",
+					errors.Newf("cross cluster replication is disabled"),
+					"You can enable cross cluster replication by running `SET CLUSTER SETTING cross_cluster_replication.enabled = true`.",
 				),
 				pgcode.ExperimentalFeature,
 			),
-			"replication.ingest.disabled",
+			"cross_cluster_replication.enabled",
 		)
 	}
 

pkg/ccl/streamingccl/streamproducer/replication_stream_test.go

@@ -176,7 +176,7 @@ func startReplication(
 	conn, err := pgx.ConnectConfig(queryCtx, pgxConfig)
 	require.NoError(t, err)
 
-	rows, err := conn.Query(queryCtx, `SET enable_experimental_stream_replication = true`)
+	rows, err := conn.Query(queryCtx, `SET CLUSTER SETTING cross_cluster_replication.enabled = true;`)
 	require.NoError(t, err)
 	rows.Close()
 

pkg/cmd/roachtest/cluster.go

@@ -1047,14 +1047,14 @@ func (c *clusterImpl) StopCockroachGracefullyOnNode(
 	gracefulOpts.RoachprodOpts.Sig = 15 // SIGTERM
 	gracefulOpts.RoachprodOpts.Wait = true
 	gracefulOpts.RoachprodOpts.MaxWait = 60
-	c.Stop(ctx, l, gracefulOpts, c.Node(node))
-	// NB: we still call Stop to make sure the process is dead when we try
-	// to restart it (or we'll catch an error from the RocksDB dir being
-	// locked). This won't happen unless run with --local due to timing.
-	c.Stop(ctx, l, option.DefaultStopOpts(), c.Node(node))
-	// TODO(tschottdorf): should return an error. I doubt that we want to
-	//  call these *testing.T-style methods on goroutines.
-	return nil
+	if err := c.StopE(ctx, l, gracefulOpts, c.Node(node)); err != nil {
+		return err
+	}
+
+	// NB: we still call Stop to make sure the process is dead when we
+	// try to restart it (in case it takes longer than `MaxWait` for it
+	// to finish).
+	return c.StopE(ctx, l, option.DefaultStopOpts(), c.Node(node))
 }
 
 // Save marks the cluster as "saved" so that it doesn't get destroyed.

pkg/cmd/roachtest/tests/cluster_to_cluster.go

@@ -611,7 +611,7 @@ func srcClusterSettings(t test.Test, db *sqlutils.SQLRunner) {
 }
 
 func destClusterSettings(t test.Test, db *sqlutils.SQLRunner) {
-	db.ExecMultiple(t, `SET enable_experimental_stream_replication = true;`)
+	db.ExecMultiple(t, `SET CLUSTER SETTING cross_cluster_replication.enabled = true;`)
 }
 
 func copyPGCertsAndMakeURL(