Merge pull request #63873 from miretskiy/backport21.1-63857

release-21.1: changefeedccl: Make `kafka_sink_config` a valid option.

pkg/ccl/changefeedccl/changefeed_test.go

@@ -2248,7 +2248,10 @@ func TestChangefeedErrors(t *testing.T) {
 		t, `param sasl_mechanism must be one of SCRAM-SHA-256, SCRAM-SHA-512, or PLAIN`,
 		`CREATE CHANGEFEED FOR foo INTO $1`, `kafka://nope/?sasl_enabled=true&sasl_mechanism=unsuppported`,
 	)
-
+	sqlDB.ExpectErr(
+		t, `client has run out of available brokers`,
+		`CREATE CHANGEFEED FOR foo INTO 'kafka://nope/' WITH kafka_sink_config='{"Flush": {"Messages": 100}}'`,
+	)
 	// The avro format doesn't support key_in_value yet.
 	sqlDB.ExpectErr(
 		t, `key_in_value is not supported with format=experimental_avro`,

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -124,4 +124,5 @@ var ChangefeedOptionExpectValues = map[string]sql.KVStringOptValidate{
 	OptInitialScan:              sql.KVStringOptRequireNoValue,
 	OptNoInitialScan:            sql.KVStringOptRequireNoValue,
 	OptProtectDataFromGCOnPause: sql.KVStringOptRequireNoValue,
+	OptKafkaSinkConfig:          sql.KVStringOptRequireValue,
 }