changefeedccl: allow per changefeed kafka quota config

[wenyihu6]

Previously, users were limited to setting a single kafka quota configuration for
cockroachdb which was then applied and restricting all changefeeds. This patch
introduces a new changefeed configuration option, allowing users to define
client id for different changefeeds, allowing users to specify different kafka
quota configurations for different changefeeds. To use it, users can specify a
unique client ID using kafka_sink_config and configure different quota
settings on kafka server based on
https://kafka.apache.org/documentation/#quotas.

CREATE CHANGEFEED FOR foo WITH kafka_sink_config='{"ClientID": "clientID1"}'

Fixes: #92290

Release note: kafka_sink_config now supports specifying a different client ID
for different changefeeds, enabling users to define distinct kafka quota
configurations for various changefeeds.

For any kafka versions >= V1_0_0_0 (KIP-190: Handle client-ids consistently
between clients and
brokers),
any string can be used as client ID. For earlier kafka versions, clientID can
only contain characters [A-Za-z0-9._-] are acceptable.

For example,

CREATE CHANGEFEED FOR ... WITH kafka_sink_config='{"ClientID": "clientID1"}'

[cockroach-teamcity]

This change is 

[wenyihu6]

Splitted the roachtest work into another PR here - #118428. We can revisit what the best way to test this work is there.

[wenyihu6]

First commit comes from #117693.

[rharding6373]

Nice work! Have a couple comments.

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

---

pkg/ccl/changefeedccl/sink_kafka.go line 838 at r2 (raw file):

	kafka.Producer.Flush.Frequency = time.Duration(c.Flush.Frequency)
	kafka.Producer.Flush.MaxMessages = c.Flush.MaxMessages
	kafka.ClientID = c.ClientID

Are there any restrictions on client IDs, e.g., length, invalid characters, that we need to validate?

---

pkg/ccl/changefeedccl/sink_test.go line 644 at r2 (raw file):

		cfg, err = getSaramaConfig(opts)
		require.NoError(t, err)
		require.NoError(t, cfg.Validate())

You should add an extra check to make sure that clientID1 is reflected in the cfg.

Is there also logic we should add toValidate to make sure the client id is sensical? See my other comment.

[wenyihu6]

pkg/ccl/changefeedccl/sink_kafka.go line 838 at r2 (raw file):

Previously, rharding6373 (Rachael Harding) wrote…

Are there any restrictions on client IDs, e.g., length, invalid characters, that we need to validate?

Sarama provides an exported validation check in its library. I opened another PR to add this check #118890.

[wenyihu6]

pkg/ccl/changefeedccl/sink_test.go line 644 at r2 (raw file):

Previously, rharding6373 (Rachael Harding) wrote…

You should add an extra check to make sure that clientID1 is reflected in the cfg.

Is there also logic we should add toValidate to make sure the client id is sensical? See my other comment.

Done.

[rharding6373]

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

[wenyihu6]

TFTRs!!

bors r=rharding6373

[craig]

Build succeeded:

• Bazel Essential CI (Cockroach)