roachtest/cdc: fix cdc/kafka-auth

From kafka 2.0 onwards, host name verification of servers is enabled by default.
ssl.endpoint.identification.algorithm defaults to `https` which validates server
host name to match the host name in the certificate. This patch fixes the
failure by pre-pending https to the sink connection URL.

Fixes: cockroachdb#118525
Release note: none

pkg/cmd/roachtest/tests/cdc.go

@@ -1044,31 +1044,31 @@ func runCDCKafkaAuth(ctx context.Context, t test.Test, c cluster.Cluster) {
 	}{
 		{
 			"create changefeed with insecure TLS transport and no auth",
-			fmt.Sprintf("%s?tls_enabled=true&insecure_tls_skip_verify=true", kafka.sinkURLTLS(ctx)),
+			fmt.Sprintf("https://%s?tls_enabled=true&insecure_tls_skip_verify=true", kafka.sinkURLTLS(ctx)),
 		},
 		{
 			"create changefeed with TLS transport and no auth",
-			fmt.Sprintf("%s?tls_enabled=true&ca_cert=%s", kafka.sinkURLTLS(ctx), testCerts.CACertBase64()),
+			fmt.Sprintf("https://%s?tls_enabled=true&ca_cert=%s", kafka.sinkURLTLS(ctx), testCerts.CACertBase64()),
 		},
 		{
 			"create changefeed with TLS transport and SASL/PLAIN (default mechanism)",
-			fmt.Sprintf("%s?tls_enabled=true&ca_cert=%s&sasl_enabled=true&sasl_user=plain&sasl_password=plain-secret", saslURL, caCert),
+			fmt.Sprintf("https://%s?tls_enabled=true&ca_cert=%s&sasl_enabled=true&sasl_user=plain&sasl_password=plain-secret", saslURL, caCert),
 		},
 		{
 			"create changefeed with TLS transport and SASL/PLAIN (explicit mechanism)",
-			fmt.Sprintf("%s?tls_enabled=true&ca_cert=%s&sasl_enabled=true&sasl_user=plain&sasl_password=plain-secret&sasl_mechanism=PLAIN", saslURL, caCert),
+			fmt.Sprintf("https://%s?tls_enabled=true&ca_cert=%s&sasl_enabled=true&sasl_user=plain&sasl_password=plain-secret&sasl_mechanism=PLAIN", saslURL, caCert),
 		},
 		{
 			"create changefeed with TLS transport and SASL/SCRAM-SHA-256",
-			fmt.Sprintf("%s?tls_enabled=true&ca_cert=%s&sasl_enabled=true&sasl_user=scram256&sasl_password=scram256-secret&sasl_mechanism=SCRAM-SHA-256", saslURL, caCert),
+			fmt.Sprintf("https://%s?tls_enabled=true&ca_cert=%s&sasl_enabled=true&sasl_user=scram256&sasl_password=scram256-secret&sasl_mechanism=SCRAM-SHA-256", saslURL, caCert),
 		},
 		{
 			"create changefeed with TLS transport and SASL/SCRAM-SHA-512",
-			fmt.Sprintf("%s?tls_enabled=true&ca_cert=%s&sasl_enabled=true&sasl_user=scram512&sasl_password=scram512-secret&sasl_mechanism=SCRAM-SHA-512", saslURL, caCert),
+			fmt.Sprintf("https://%s?tls_enabled=true&ca_cert=%s&sasl_enabled=true&sasl_user=scram512&sasl_password=scram512-secret&sasl_mechanism=SCRAM-SHA-512", saslURL, caCert),
 		},
 		{
 			"create changefeed with confluent-cloud scheme",
-			fmt.Sprintf("%s&api_key=plain&api_secret=plain-secret", kafka.sinkURLAsConfluentCloudUrl(ctx)),
+			fmt.Sprintf("https://%s&api_key=plain&api_secret=plain-secret", kafka.sinkURLAsConfluentCloudUrl(ctx)),
 		},
 	}