diff --git a/internal/kafka/command_topic_create.go b/internal/kafka/command_topic_create.go index 462ead4d53..93544eb324 100644 --- a/internal/kafka/command_topic_create.go +++ b/internal/kafka/command_topic_create.go @@ -19,6 +19,11 @@ import ( "github.com/confluentinc/cli/v4/pkg/utils" ) +// confluent.*.association values are JSON strings. These keys are passed to GetMapFromArray as +// jsonValueKeys, so their values are located by JSON validity, validated as JSON, and stored +// verbatim (skipping special-character un-escaping). +var associationConfigs = []string{"confluent.key.association", "confluent.value.association"} + func (c *command) newCreateCommand() *cobra.Command { cmd := &cobra.Command{ Use: "create ", @@ -35,7 +40,7 @@ func (c *command) newCreateCommand() *cobra.Command { } cmd.Flags().Uint32("partitions", 0, "Number of topic partitions.") - cmd.Flags().StringSlice("config", nil, `A comma-separated list of configuration overrides ("key=value") for the topic being created.`) + pcmd.AddTopicConfigFlag(cmd) pcmd.AddEndpointFlag(cmd, c.AuthenticatedCLICommand) pcmd.AddDryRunFlag(cmd) cmd.Flags().Bool("if-not-exists", false, "Exit gracefully if topic already exists.") @@ -54,12 +59,11 @@ func (c *command) create(cmd *cobra.Command, args []string) error { return err } - configs, err := cmd.Flags().GetStringSlice("config") + configs, err := cmd.Flags().GetStringArray("config") if err != nil { return err } - - configMap, err := properties.ConfigFlagToMap(configs) + configMap, err := properties.GetMapFromArray(configs, associationConfigs...) if err != nil { return err } diff --git a/internal/kafka/command_topic_update.go b/internal/kafka/command_topic_update.go index 68374b9785..71584165f3 100644 --- a/internal/kafka/command_topic_update.go +++ b/internal/kafka/command_topic_update.go @@ -44,7 +44,7 @@ func (c *command) newUpdateCommand() *cobra.Command { Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireNonAPIKeyCloudLogin}, } - pcmd.AddConfigFlag(cmd) + pcmd.AddTopicConfigFlag(cmd) pcmd.AddEndpointFlag(cmd, c.AuthenticatedCLICommand) pcmd.AddDryRunFlag(cmd) pcmd.AddClusterFlag(cmd, c.AuthenticatedCLICommand) @@ -60,11 +60,11 @@ func (c *command) newUpdateCommand() *cobra.Command { func (c *command) update(cmd *cobra.Command, args []string) error { topicName := args[0] - configs, err := cmd.Flags().GetStringSlice("config") + configs, err := cmd.Flags().GetStringArray("config") if err != nil { return err } - configMap, err := properties.GetMap(configs) + configMap, err := properties.GetMapFromArray(configs, associationConfigs...) if err != nil { return err } diff --git a/pkg/cmd/flags.go b/pkg/cmd/flags.go index 1c10e9bd21..f2aec1f69b 100644 --- a/pkg/cmd/flags.go +++ b/pkg/cmd/flags.go @@ -150,6 +150,13 @@ func AddConfigFlag(cmd *cobra.Command) { cmd.Flags().StringSlice("config", []string{}, `A comma-separated list of "key=value" pairs, or path to a configuration file containing a newline-separated list of "key=value" pairs.`) } +// AddTopicConfigFlag registers the topic `--config` flag as a StringArray so that +// pflag does not CSV-split the value; this lets a "key=value" pair carry a JSON value (APIE-1106) +// Callers must read it with GetStringArray and parse it with properties.GetMapFromArray. +func AddTopicConfigFlag(cmd *cobra.Command) { + cmd.Flags().StringArray("config", nil, `A comma-separated list of "key=value" pairs, or path to a configuration file containing a newline-separated list of "key=value" pairs.`) +} + func AddContextFlag(cmd *cobra.Command, command *CLICommand) { cmd.Flags().String("context", "", "CLI context name.") diff --git a/pkg/properties/properties.go b/pkg/properties/properties.go index 730cc9ca70..baa42a64d5 100644 --- a/pkg/properties/properties.go +++ b/pkg/properties/properties.go @@ -2,8 +2,10 @@ package properties import ( "bytes" + "encoding/json" "fmt" "os" + "slices" "sort" "strings" @@ -20,17 +22,17 @@ func GetMap(config []string) (map[string]string, error) { } // fileToMap reads key=value pairs from a properties file, ignoring comments and empty lines. -func fileToMap(filename string) (map[string]string, error) { +func fileToMap(filename string, rawValueKeys ...string) (map[string]string, error) { buf, err := os.ReadFile(filename) if err != nil { return nil, err } - return ConfigSliceToMap(ParseLines(string(buf))) + return ConfigSliceToMap(ParseLines(string(buf)), rawValueKeys...) } // ConfigSliceToMap converts a list of key=value strings into a map. -func ConfigSliceToMap(configs []string) (map[string]string, error) { +func ConfigSliceToMap(configs []string, rawValueKeys ...string) (map[string]string, error) { m := make(map[string]string) for _, config := range configs { @@ -39,7 +41,12 @@ func ConfigSliceToMap(configs []string) (map[string]string, error) { return nil, fmt.Errorf(`failed to parse "key=value" pattern from configuration: %s`, config) } - m[x[0]] = replaceSpecialCharacters(x[1]) + // rawValueKeys are stored as-is, all other values get un-escaped. + if slices.Contains(rawValueKeys, x[0]) { + m[x[0]] = x[1] + } else { + m[x[0]] = replaceSpecialCharacters(x[1]) + } } return m, nil @@ -83,6 +90,82 @@ func ConfigFlagToMap(configs []string) (map[string]string, error) { return m, nil } +// GetMapFromArray reads configuration from a configuration file or from a StringArray. It supports values containing commas. +// Values whose keys are listed in jsonValueKeys are treated as JSON. +// Use this instead of GetMap for flags registered with cmd.AddTopicConfigFlag. +func GetMapFromArray(config []string, jsonValueKeys ...string) (map[string]string, error) { + if len(config) == 1 && utils.FileExists(config[0]) { + return fileToMap(config[0], jsonValueKeys...) + } + + return configArrayToMap(config, jsonValueKeys...) +} + +// configArrayToMap parses raw config elements into a map, each element is split on commas into "key=value" pairs. +// JSON config values as indicated by jsonValueKeys are preserved and validated. +func configArrayToMap(configs []string, jsonValueKeys ...string) (map[string]string, error) { + m := make(map[string]string) + + for _, config := range configs { + for _, pair := range splitConfigPairs(config, jsonValueKeys) { + x := strings.SplitN(pair, "=", 2) + if len(x) < 2 { + return nil, fmt.Errorf(`failed to parse "key=value" pattern from configuration: %s`, pair) + } + + if slices.Contains(jsonValueKeys, x[0]) { + if !json.Valid([]byte(strings.TrimSpace(x[1]))) { + return nil, fmt.Errorf(`failed to parse JSON value for configuration "%s": %s`, x[0], x[1]) + } + m[x[0]] = x[1] + } else { + m[x[0]] = replaceSpecialCharacters(x[1]) + } + } + } + + return m, nil +} + +// splitConfigPairs splits config into raw "key=value" pair strings, honoring +// comma-separated values and JSON values for jsonValueKeys. +func splitConfigPairs(config string, jsonValueKeys []string) []string { + var pairs []string + + current := "" + for _, fragment := range strings.Split(config, ",") { + switch { + case current == "": + current = fragment + case strings.Contains(fragment, "=") && pairComplete(current, jsonValueKeys): + pairs = append(pairs, current) + current = fragment + default: + // A comma inside the current value: glue the fragment back on. + current += "," + fragment + } + } + if current != "" { + pairs = append(pairs, current) + } + + return pairs +} + +// pairComplete reports whether an accumulated "key=value" is complete. For a jsonValueKey the value must be valid JSON. +func pairComplete(pair string, jsonValueKeys []string) bool { + x := strings.SplitN(pair, "=", 2) + if len(x) < 2 { + return true + } + + if slices.Contains(jsonValueKeys, x[0]) { + return json.Valid([]byte(strings.TrimSpace(x[1]))) + } + + return true +} + func CreateKeyValuePairs(m map[string]string) string { // Sort by keys so the output order is predictable which is helpful for testing. keys := make([]string, 0, len(m)) diff --git a/pkg/properties/properties_test.go b/pkg/properties/properties_test.go index e25f795dfe..a2548f962c 100644 --- a/pkg/properties/properties_test.go +++ b/pkg/properties/properties_test.go @@ -1,6 +1,8 @@ package properties import ( + "os" + "path/filepath" "testing" "github.com/stretchr/testify/require" @@ -124,3 +126,76 @@ func TestCreateKeyValuePairsKeysWithDotsAndSorts(t *testing.T) { m["connection.mode"] = "OUTBOUND" require.Equal(t, "\"connection.mode\"=\"OUTBOUND\"\n\"link.mode\"=\"BIDIRECTIONAL\"\n", CreateKeyValuePairs(m)) } + +// Regression: behavior must match the StringSlice path for non-JSON configs. StringArray delivers a +// comma-list as ONE element (pflag does not split), so the parser must split it itself. +func TestConfigArrayToMap_CommaSeparatedList(t *testing.T) { + m, err := GetMapFromArray([]string{"cleanup.policy=compact,compression.type=gzip"}) + require.NoError(t, err) + require.Equal(t, map[string]string{"cleanup.policy": "compact", "compression.type": "gzip"}, m) +} + +func TestConfigArrayToMap_ValueWithComma(t *testing.T) { + m, err := GetMapFromArray([]string{"cleanup.policy=delete,compact"}) + require.NoError(t, err) + require.Equal(t, map[string]string{"cleanup.policy": "delete,compact"}, m) +} + +func TestConfigArrayToMap_MultipleFlags(t *testing.T) { + m, err := GetMapFromArray([]string{"a=1", "b=2"}) + require.NoError(t, err) + require.Equal(t, map[string]string{"a": "1", "b": "2"}, m) +} + +func TestConfigArrayToMap_Override(t *testing.T) { + m, err := GetMapFromArray([]string{"retention.ms=1", "retention.ms=2"}) + require.NoError(t, err) + require.Equal(t, map[string]string{"retention.ms": "2"}, m) +} + +func TestConfigArrayToMap_ValueWithEquals(t *testing.T) { + m, err := GetMapFromArray([]string{"key=a=b"}) + require.NoError(t, err) + require.Equal(t, map[string]string{"key": "a=b"}, m) +} + +func TestConfigArrayToMap_UnescapesNonRawValues(t *testing.T) { + m, err := GetMapFromArray([]string{`foo=a\nb`}) + require.NoError(t, err) + require.Equal(t, map[string]string{"foo": "a\nb"}, m) +} + +func TestConfigArrayToMap_RawValueKeyPreservesJSON(t *testing.T) { + json := `{"schema":"{\"type\":\"record\",\"name\":\"TestRecord\",\"doc\":\"Basic test.\\na=b.\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"]}]}"}` + m, err := GetMapFromArray([]string{"confluent.value.association=" + json}, "confluent.value.association") + require.NoError(t, err) + require.Equal(t, json, m["confluent.value.association"]) +} + +func TestConfigArrayToMap_JSONWithCommasThenNormalConfig(t *testing.T) { + m, err := GetMapFromArray( + []string{`confluent.key.association={"subject":"x","lifecycle":"STRONG"},retention.ms=500`}, + "confluent.key.association", + ) + require.NoError(t, err) + require.Equal(t, `{"subject":"x","lifecycle":"STRONG"}`, m["confluent.key.association"]) + require.Equal(t, "500", m["retention.ms"]) +} + +func TestConfigArrayToMap_MalformedJSONReturnsError(t *testing.T) { + _, err := GetMapFromArray([]string{`confluent.value.association={"broken`}, "confluent.value.association") + require.Error(t, err) +} + +// The file path is unchanged and shared with GetMap. +func TestGetMapFromArray_FilePreservesJSON(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "topic.properties") + // The file path stores jsonValueKeys verbatim: a JSON value containing "\n" and "=" must not be un-escaped. + json := `{"schema":"{\"type\":\"record\",\"name\":\"TestRecord\",\"doc\":\"Basic test.\\na=b.\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"]}]}"}` + require.NoError(t, os.WriteFile(path, []byte("confluent.value.association="+json+"\n"), 0o600)) + + m, err := GetMapFromArray([]string{path}, "confluent.value.association") + require.NoError(t, err) + require.Equal(t, json, m["confluent.value.association"]) +} diff --git a/test/fixtures/input/kafka/topic/topic-config-json.properties b/test/fixtures/input/kafka/topic/topic-config-json.properties new file mode 100644 index 0000000000..0a8bb7e825 --- /dev/null +++ b/test/fixtures/input/kafka/topic/topic-config-json.properties @@ -0,0 +1,2 @@ +confluent.key.association={"subject":"kafkaCLISubject"} +confluent.value.association={"schema":"{\"type\":\"record\",\"name\":\"TestRecord\",\"doc\":\"Basic test.\\na=b.\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"]}]}"} diff --git a/test/fixtures/input/kafka/topic/topic-config.properties b/test/fixtures/input/kafka/topic/topic-config.properties new file mode 100644 index 0000000000..71b4ed7ad9 --- /dev/null +++ b/test/fixtures/input/kafka/topic/topic-config.properties @@ -0,0 +1,2 @@ +retention.ms=259200000 +compression.type=gzip diff --git a/test/fixtures/output/kafka/topic/create-help.golden b/test/fixtures/output/kafka/topic/create-help.golden index b7b983381b..7484a5cb29 100644 --- a/test/fixtures/output/kafka/topic/create-help.golden +++ b/test/fixtures/output/kafka/topic/create-help.golden @@ -10,7 +10,7 @@ Create a topic named "my_topic" with default options. Flags: --partitions uint32 Number of topic partitions. - --config strings A comma-separated list of configuration overrides ("key=value") for the topic being created. + --config stringArray A comma-separated list of "key=value" pairs, or path to a configuration file containing a newline-separated list of "key=value" pairs. --kafka-endpoint string Endpoint to be used for this Kafka cluster. --dry-run Run the command without committing changes. --if-not-exists Exit gracefully if topic already exists. diff --git a/test/fixtures/output/kafka/topic/create.golden b/test/fixtures/output/kafka/topic/create.golden index 2c6d0a0b8f..5779c62a94 100644 --- a/test/fixtures/output/kafka/topic/create.golden +++ b/test/fixtures/output/kafka/topic/create.golden @@ -9,7 +9,7 @@ Create a topic named "my_topic" with default options. Flags: --partitions uint32 Number of topic partitions. - --config strings A comma-separated list of configuration overrides ("key=value") for the topic being created. + --config stringArray A comma-separated list of "key=value" pairs, or path to a configuration file containing a newline-separated list of "key=value" pairs. --kafka-endpoint string Endpoint to be used for this Kafka cluster. --dry-run Run the command without committing changes. --if-not-exists Exit gracefully if topic already exists. diff --git a/test/fixtures/output/kafka/topic/update-help.golden b/test/fixtures/output/kafka/topic/update-help.golden index 4e111bba72..5f58a0cc23 100644 --- a/test/fixtures/output/kafka/topic/update-help.golden +++ b/test/fixtures/output/kafka/topic/update-help.golden @@ -9,7 +9,7 @@ Modify the "my_topic" topic to have a retention period of 3 days (259200000 mill $ confluent kafka topic update my_topic --config retention.ms=259200000 Flags: - --config strings REQUIRED: A comma-separated list of "key=value" pairs, or path to a configuration file containing a newline-separated list of "key=value" pairs. + --config stringArray REQUIRED: A comma-separated list of "key=value" pairs, or path to a configuration file containing a newline-separated list of "key=value" pairs. --kafka-endpoint string Endpoint to be used for this Kafka cluster. --dry-run Run the command without committing changes. --cluster string Kafka cluster ID. diff --git a/test/fixtures/output/kafka/topic/update-json-config.golden b/test/fixtures/output/kafka/topic/update-json-config.golden new file mode 100644 index 0000000000..40f7030907 --- /dev/null +++ b/test/fixtures/output/kafka/topic/update-json-config.golden @@ -0,0 +1,6 @@ +Updated the following configuration values for topic "topic-exist-rest": + Name | Value | Read-Only +------------------------------+-------------------------------------------------------------------------------------+------------ + confluent.key.association | {"subject":"kafkaCLISubject"} | false + confluent.value.association | {"schema":"{\"type\":\"record\",\"name\":\"TestRecord\",\"doc\":\"Basic | false + | test.\\na=b.\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"]}]}"} | diff --git a/test/kafka_test.go b/test/kafka_test.go index 3e146e84ae..521c364bd8 100644 --- a/test/kafka_test.go +++ b/test/kafka_test.go @@ -152,6 +152,11 @@ func (s *CLITestSuite) TestKafka() { {args: "kafka topic create", login: "cloud", useKafka: "lkc-create-topic", fixture: "kafka/topic/create.golden", exitCode: 1}, {args: "kafka topic create topic1", useKafka: "lkc-create-topic", fixture: "kafka/topic/create-success.golden"}, {args: "kafka topic create topic1 --dry-run", useKafka: "lkc-create-topic", fixture: "kafka/topic/create-success.golden"}, + {args: "kafka topic create topic1 --config test/fixtures/input/kafka/topic/topic-config.properties", useKafka: "lkc-create-topic", fixture: "kafka/topic/create-success.golden"}, + {args: "kafka topic create topic1 --config test/fixtures/input/kafka/topic/topic-config-json.properties", useKafka: "lkc-create-topic", fixture: "kafka/topic/create-success.golden"}, + {args: `kafka topic create topic1 --config 'confluent.key.association={"subject":"kafkaCLISubject"}' --config 'confluent.value.association={"schema":"{\"type\":\"record\",\"name\":\"TestRecord\",\"doc\":\"Basic test.\\na=b.\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"]}]}"}'`, useKafka: "lkc-create-topic", fixture: "kafka/topic/create-success.golden", name: "create topic with inline JSON association configs"}, + {args: `kafka topic create topic1 --config 'confluent.key.association={"subject":"kafkaCLISubject"},confluent.value.association={"schema":"{\"type\":\"record\",\"name\":\"TestRecord\",\"doc\":\"Basic test.\\na=b.\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"]}]}"}'`, useKafka: "lkc-create-topic", fixture: "kafka/topic/create-success.golden", name: "create topic with inline JSON association configs 2"}, + {args: `kafka topic create topic1 --config retention.ms=1 --config 'confluent.value.association={"schema":"{\"type\":\"record\",\"name\":\"TestRecord\",\"doc\":\"Basic test.\\na=b.\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"]}]}"}'`, useKafka: "lkc-create-topic", fixture: "kafka/topic/create-success.golden", name: "create topic with a JSON config and a plain config"}, {args: "kafka topic create topic-exist", login: "cloud", useKafka: "lkc-create-topic", fixture: "kafka/topic/create-dup-topic.golden", exitCode: 1}, {args: "kafka topic create topic-exceed-limit --partitions 9001", login: "cloud", useKafka: "lkc-create-topic", fixture: "kafka/topic/create-limit-topic.golden", exitCode: 1}, @@ -178,6 +183,9 @@ func (s *CLITestSuite) TestKafka() { {args: "kafka topic update topic-exist-rest --config retention.ms=1,compression.type=gzip -o json", useKafka: "lkc-describe-topic", fixture: "kafka/topic/update-success-rest-json.golden"}, {args: "kafka topic update topic-exist-rest --config retention.ms=1,compression.type=gzip -o yaml", useKafka: "lkc-describe-topic", fixture: "kafka/topic/update-success-rest-yaml.golden"}, {args: "kafka topic update topic-exist-rest --config num.partitions=6", useKafka: "lkc-describe-topic", fixture: "kafka/topic/update-success-rest-partitions-count.golden"}, + {args: "kafka topic update topic-exist-rest --config test/fixtures/input/kafka/topic/topic-config.properties", useKafka: "lkc-describe-topic", fixture: "kafka/topic/update-success-rest.golden"}, + {args: "kafka topic update topic-exist-rest --config test/fixtures/input/kafka/topic/topic-config-json.properties", useKafka: "lkc-describe-topic", fixture: "kafka/topic/update-json-config.golden"}, + {args: `kafka topic update topic-exist-rest --config 'confluent.key.association={"subject":"kafkaCLISubject"}' --config 'confluent.value.association={"schema":"{\"type\":\"record\",\"name\":\"TestRecord\",\"doc\":\"Basic test.\\na=b.\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"]}]}"}'`, useKafka: "lkc-describe-topic", fixture: "kafka/topic/update-json-config.golden", name: "update topic with inline JSON association configs"}, } if runtime.GOOS != "windows" { diff --git a/test/test-server/kafka_rest_router.go b/test/test-server/kafka_rest_router.go index 06e2c204f2..eb139463a0 100644 --- a/test/test-server/kafka_rest_router.go +++ b/test/test-server/kafka_rest_router.go @@ -5,6 +5,7 @@ import ( "fmt" "io" "net/http" + "slices" "strconv" "strings" "testing" @@ -27,6 +28,24 @@ const ( shareGroupID1 = "share-group-1" ) +// allowedTopicConfigNames are the topic configs the mock Kafka REST server accepts on +// topic create and update. confluent.key/value.association hold JSON string values (APIE-1106). +var allowedTopicConfigNames = []string{ + "retention.ms", + "compression.type", + "confluent.key.association", + "confluent.value.association", +} + +// expectedAssociationConfigValues are the exact JSON values in +// test/fixtures/input/kafka/topic/topic-config-json.properties, keyed by config name. The mock +// asserts the CLI forwards each byte-for-byte, proving GetMap read the file without un-escaping +// the value (APIE-1106). +var expectedAssociationConfigValues = map[string]string{ + "confluent.key.association": `{"subject":"kafkaCLISubject"}`, + "confluent.value.association": `{"schema":"{\"type\":\"record\",\"name\":\"TestRecord\",\"doc\":\"Basic test.\\na=b.\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"]}]}"}`, +} + type route struct { path string handler func(t *testing.T) http.HandlerFunc @@ -217,7 +236,7 @@ func handleKafkaRestTopics(t *testing.T) http.HandlerFunc { } // check configs for _, config := range requestData.Configs { - if config.Name != "retention.ms" && config.Name != "compression.type" { + if !slices.Contains(allowedTopicConfigNames, config.Name) { require.NoError(t, writeErrorResponse(w, http.StatusBadRequest, 40002, fmt.Sprintf("Unknown topic config name: %s", config.Name))) return } else if config.Name == "retention.ms" { @@ -228,6 +247,10 @@ func handleKafkaRestTopics(t *testing.T) http.HandlerFunc { require.NoError(t, writeErrorResponse(w, http.StatusBadRequest, 40002, fmt.Sprintf("Invalid value %s for configuration retention.ms: Not a number of type LONG", *config.Value))) return } + } else if expected, ok := expectedAssociationConfigValues[config.Name]; ok { + // APIE-1106: the JSON value must arrive byte-for-byte as written in the fixture file. + require.NotNil(t, config.Value) + require.Equal(t, expected, *config.Value) } // TODO: check for compression.type } @@ -303,6 +326,16 @@ func handleKafkaRestTopicConfigs(t *testing.T) http.HandlerFunc { Name: "retention.ms", Value: ptrString("1"), }, + // APIE-1106: echo the association configs so a successful update lists their + // stored JSON values (same source of truth as the alter-handler assertion). + { + Name: "confluent.key.association", + Value: ptrString(expectedAssociationConfigValues["confluent.key.association"]), + }, + { + Name: "confluent.value.association", + Value: ptrString(expectedAssociationConfigValues["confluent.value.association"]), + }, }, } reply, err := json.Marshal(topicConfigList) @@ -514,7 +547,7 @@ func handleKafkaRestConfigsAlter(t *testing.T) http.HandlerFunc { // Check Alter Args if valid for _, config := range requestData.Data { - if config.Name != "retention.ms" && config.Name != "compression.type" { // should be either retention.ms or compression.type + if !slices.Contains(allowedTopicConfigNames, config.Name) { require.NoError(t, writeErrorResponse(w, http.StatusNotFound, 404, fmt.Sprintf("Config %s cannot be found for TOPIC topic-exist in cluster cluster-1.", config.Name))) return } else if config.Name == "retention.ms" { @@ -525,6 +558,10 @@ func handleKafkaRestConfigsAlter(t *testing.T) http.HandlerFunc { require.NoError(t, writeErrorResponse(w, http.StatusBadRequest, 40002, fmt.Sprintf("Invalid config value for resource ConfigResource(type=TOPIC, name='topic-exist'): Invalid value %s for configuration retention.ms: Not a number of type LONG", *config.Value))) return } + } else if expected, ok := expectedAssociationConfigValues[config.Name]; ok { + // APIE-1106: the JSON value must arrive byte-for-byte as written in the fixture file. + require.NotNil(t, config.Value) + require.Equal(t, expected, *config.Value) } // TODO check for compression.type values }