Merge cockroachdb#153902

153902: logical: deflake TestRandomStream r=jeffswenson a=jeffswenson

TestRandomStream passes a non-external URI. This causes the external connection polling logic added by cockroachdb#149261 to return an error which results in the processor shutting down right away. The only reason the test mostly works is the tear down takes some time and we are usually able to replicate a few rows before shutdown completes.

Release note: none
Fixes: cockroachdb#153666
Fixes: cockroachdb#152435

Co-authored-by: Jeff Swenson <[email protected]>

Author: craig[bot]

pkg/crosscluster/logical/BUILD.bazel

@@ -34,7 +34,6 @@ go_library(
         "//pkg/ccl/changefeedccl/changefeedbase",
         "//pkg/ccl/utilccl",
         "//pkg/cloud",
-        "//pkg/cloud/externalconn",
         "//pkg/clusterversion",
         "//pkg/crosscluster",
         "//pkg/crosscluster/physical",

pkg/crosscluster/logical/logical_replication_job.go

@@ -8,11 +8,9 @@ package logical
 import (
 	"context"
 	"fmt"
-	"net/url"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
-	"github.com/cockroachdb/cockroach/pkg/cloud/externalconn"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/physical"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/replicationutils"
@@ -1092,21 +1090,17 @@ func getRetryPolicy(knobs *sql.StreamingTestingKnobs) retry.Options {
 func resolveDest(
 	ctx context.Context, execCfg *sql.ExecutorConfig, sourceURI string,
 ) (string, error) {
-	u, err := url.Parse(sourceURI)
+	configUri, err := streamclient.ParseConfigUri(sourceURI)
 	if err != nil {
 		return "", err
 	}
 
-	resolved := ""
-	err = execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
-		conn, err := externalconn.LoadExternalConnection(ctx, u.Host, txn)
-		if err != nil {
-			return err
-		}
-		resolved = conn.UnredactedConnectionStatement()
-		return nil
-	})
-	return resolved, err
+	clusterUri, err := configUri.AsClusterUri(ctx, execCfg.InternalDB)
+	if err != nil {
+		return "", err
+	}
+
+	return clusterUri.Serialize(), nil
 }
 
 func reloadDest(ctx context.Context, id jobspb.JobID, execCfg *sql.ExecutorConfig) (string, error) {