roachtest/cdc: fix cdc/kafka-auth

[wenyihu6]

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: #118525
Release note: none

[cockroach-teamcity]

This change is 

[rharding6373]

Thanks for tracking this down!

Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @jayshrivastava and @wenyihu6)

---

-- commits line 6 at r1:
Is this only relevant to the roachtests? Is there anything that needs to be updated or recommended in our kafka sink uri documentation? https://www.cockroachlabs.com/docs/v23.2/changefeed-sinks#kafka

[wenyihu6]

-- commits line 6 at r1:

Previously, rharding6373 (Rachael Harding) wrote…

Is this only relevant to the roachtests? Is there anything that needs to be updated or recommended in our kafka sink uri documentation? https://www.cockroachlabs.com/docs/v23.2/changefeed-sinks#kafka

I think this is only relevant to the roachtests. Customers should have kafka nodes properly set up ssl.endpoint.identification.algorithm is a kafka configuration like what we did here

cockroach/pkg/cmd/roachtest/tests/cdc.go

Line 1908 in bf5ffc4

kafkaOauthConfigTmpl = `

. But I'm still confused about why other kafka roachtests didn't fail and about auth in general. It was a pure luck that I got this fixed haha. Hope Steven could shed some light tomorrow.

[stevendanna]

Nice work digging into this so far.

This change isn't quite what we want. You are right that the underlying problem is related to host name verification.

However, what your change here is doing is changing the changefeed URL we use to create the changefeed from one that starts with kafka:// to one that starts with https://kafka://. While I haven't confirmed this, what I suspect is then happening is that when we create a changefeed the "scheme" of the URL is https:// with a host of kafka://rest_of_url. As a result, a cloudstorage sink is started. Now, that cloud storage sink will fail, but not during planning/job creation and this test assumes that any auth failure will happen via job creation. The kafka sink fails during planning because it implements the Dial method which attempts to connect to the sink. The cloud storage sink simply return nil from Dial. As a result, the test passes, but not for the right reason.

I think we have a couple of options on how to fix the underlying error. (1) change the kafka configuration used for controller to broker connections or (2) create a TLS cert with our fully qualified domain name in the DNS alternative names. We can discuss both in detail today.

[wenyihu6]

Closing in favor of #119077 - this old PR appeared to fix the problem but is actually using a cloudstorage sink rather than a kafka sink.

cockroach/pkg/ccl/changefeedccl/sink_cloudstorage.go

Lines 52 to 54 in e9a6d3d

	// During the deprecation period, we need to keep parsing these as cloudstorage for backwards
	// compatibility. Afterwards we'll either remove them or move them to webhook.
	case changefeedbase.DeprecatedSinkSchemeHTTP, changefeedbase.DeprecatedSinkSchemeHTTPS: