[APIE-1106] Update kafka topic create/update to support JSON-valued Configs

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 <topic>",
@@ -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
 	}

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
 	}

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.")
 

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))

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"])
+}

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\"]}]}"}

test/fixtures/input/kafka/topic/topic-config.properties

@@ -0,0 +1,2 @@
+retention.ms=259200000
+compression.type=gzip

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.

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.

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.

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\"]}]}"} |            

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" {