Merge #126276 #126455

126276: roachtest: fix crash on cluster creation failure r=srosenberg,darrylwong,herkolategan a=andreimatei

Before this patch, we'd segfault if cluster creation fails and we're running with --debug-always.

This patch is a small protection against the problem, but I don't think it's the right fix. The structure of the code makes similar bugs very likely to introduce -- I've made the same mistake myself. The issue is that, since #73706, the point where the cluster creation fails is disconnected from the point where that error is handled. It's very easy for code in between to assume that there is a cluster. Besides the bug being fixed here, there are some other awkward protections against a nil cluster around that seem surprising.

Epic: None
Release note: None

126455: crosscluster/logical: support external:// r=dt a=dt

Release note: none.
Epic: none.

Co-authored-by: Andrei Matei <[email protected]>
Co-authored-by: David Taylor <[email protected]>

pkg/ccl/crosscluster/logical/logical_replication_job.go

@@ -11,7 +11,6 @@ package logical
 import (
 	"context"
 	"fmt"
-	"net/url"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
@@ -118,12 +117,13 @@ func (r *logicalReplicationResumer) ingest(
 		replicatedTimeAtStart = progress.ReplicatedTime
 	)
 
-	streamAddr, err := url.Parse(payload.TargetClusterConnStr)
-	if err != nil {
-		return err
-	}
+	client, err := streamclient.NewStreamClient(ctx,
+		crosscluster.StreamAddress(payload.TargetClusterConnStr),
+		jobExecCtx.ExecCfg().InternalDB,
+		streamclient.WithStreamID(streampb.StreamID(streamID)),
+		streamclient.WithLogical(),
+	)
 
-	client, err := streamclient.NewPartitionedStreamClient(ctx, streamAddr, streamclient.WithStreamID(streampb.StreamID(streamID)), streamclient.WithLogical())
 	if err != nil {
 		return err
 	}

pkg/ccl/crosscluster/logical/logical_replication_job_test.go

@@ -12,6 +12,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"net/url"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -149,6 +150,14 @@ func TestLogicalStreamIngestionJob(t *testing.T) {
 	dbBURL, cleanupB := s.PGUrl(t, serverutils.DBName("b"))
 	defer cleanupB()
 
+	// Swap one of the URLs to external:// to verify this indirection works.
+	// TODO(dt): this create should support placeholder for URI.
+	dbB.Exec(t, "CREATE EXTERNAL CONNECTION a AS '"+dbAURL.String()+"'")
+	dbAURL = url.URL{
+		Scheme: "external",
+		Host:   "a",
+	}
+
 	var (
 		jobAID jobspb.JobID
 		jobBID jobspb.JobID

pkg/cmd/roachtest/test_runner.go

@@ -749,7 +749,7 @@ func (r *testRunner) runWorker(
 		// If DebugKeepAlways is set, mark it as a saved cluster, so it isn't
 		// cleaned up. Do it now instead of at the end as the test may be interrupted
 		// with ctrl c before we get there.
-		if clustersOpt.debugMode == DebugKeepAlways {
+		if c != nil && clustersOpt.debugMode == DebugKeepAlways {
 			c.Save(ctx, "cluster saved since --debug-always set", l)
 		}