Merge pull request #63872 from miretskiy/backport20.2-63857

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

pkg/ccl/changefeedccl/changefeed_test.go

@@ -2083,7 +2083,10 @@ func TestChangefeedErrors(t *testing.T) {
 		t, `sasl_enabled must be enabled if a SASL password is provided`,
 		`CREATE CHANGEFEED FOR foo INTO $1`, `kafka://nope/?sasl_password=a`,
 	)
-
+	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

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