From 0526262ff2f077988ba0c39f87965fdd3d1ecc5c Mon Sep 17 00:00:00 2001
From: Yifei Yuan <yyuan@confluent.io>
Date: Mon, 29 Jun 2026 19:56:11 -0400
Subject: [PATCH 01/10] Update GetMap to preserves configs JSON values

---
 internal/kafka/command_topic_create.go |  9 +++++---
 internal/kafka/command_topic_update.go |  2 +-
 pkg/properties/properties.go           | 29 ++++++++++++++++++--------
 3 files changed, 27 insertions(+), 13 deletions(-)

diff --git a/internal/kafka/command_topic_create.go b/internal/kafka/command_topic_create.go
index 462ead4d53..aa31422883 100644
--- a/internal/kafka/command_topic_create.go
+++ b/internal/kafka/command_topic_create.go
@@ -19,6 +19,10 @@ import (
 	"github.com/confluentinc/cli/v4/pkg/utils"
 )
 
+// confluent.*.association values are JSON strings. They must be stored verbatim,
+// so keys are passed as rawValueKeys to skip special-character un-escaping.
+var rawValueTopicConfigs = []string{"confluent.key.association", "confluent.value.association"}
+
 func (c *command) newCreateCommand() *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "create <topic>",
@@ -35,7 +39,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.AddConfigFlag(cmd)
 	pcmd.AddEndpointFlag(cmd, c.AuthenticatedCLICommand)
 	pcmd.AddDryRunFlag(cmd)
 	cmd.Flags().Bool("if-not-exists", false, "Exit gracefully if topic already exists.")
@@ -58,8 +62,7 @@ func (c *command) create(cmd *cobra.Command, args []string) error {
 	if err != nil {
 		return err
 	}
-
-	configMap, err := properties.ConfigFlagToMap(configs)
+	configMap, err := properties.GetMap(configs, rawValueTopicConfigs...)
 	if err != nil {
 		return err
 	}
diff --git a/internal/kafka/command_topic_update.go b/internal/kafka/command_topic_update.go
index 68374b9785..24a007d5d5 100644
--- a/internal/kafka/command_topic_update.go
+++ b/internal/kafka/command_topic_update.go
@@ -64,7 +64,7 @@ func (c *command) update(cmd *cobra.Command, args []string) error {
 	if err != nil {
 		return err
 	}
-	configMap, err := properties.GetMap(configs)
+	configMap, err := properties.GetMap(configs, rawValueTopicConfigs...)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/properties/properties.go b/pkg/properties/properties.go
index 730cc9ca70..b23f601121 100644
--- a/pkg/properties/properties.go
+++ b/pkg/properties/properties.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"fmt"
 	"os"
+	"slices"
 	"sort"
 	"strings"
 
@@ -11,26 +12,27 @@ import (
 )
 
 // GetMap reads newline-separated configuration files or comma-separated lists of key=value pairs, and supports configuration values containing commas.
-func GetMap(config []string) (map[string]string, error) {
+// Values whose keys are listed in rawValueKeys are stored verbatim, skipping special characters un-escaping, to preserve values such as JSON strings unchanged.
+func GetMap(config []string, rawValueKeys ...string) (map[string]string, error) {
 	if len(config) == 1 && utils.FileExists(config[0]) {
-		return fileToMap(config[0])
+		return fileToMap(config[0], rawValueKeys...)
 	}
 
-	return ConfigFlagToMap(config)
+	return ConfigFlagToMap(config, rawValueKeys...)
 }
 
 // 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
@@ -62,14 +69,18 @@ func ParseLines(content string) []string {
 }
 
 // ConfigFlagToMap reads key=values pairs from the --config flag and supports configuration values containing commas.
-func ConfigFlagToMap(configs []string) (map[string]string, error) {
+func ConfigFlagToMap(configs []string, rawValueKeys ...string) (map[string]string, error) {
 	m := make(map[string]string)
 
 	for i := len(configs) - 1; i >= 0; i-- {
 		if strings.Contains(configs[i], "=") {
 			x := strings.SplitN(configs[i], "=", 2)
 			if _, ok := m[x[0]]; !ok {
-				m[x[0]] = replaceSpecialCharacters(x[1])
+				if slices.Contains(rawValueKeys, x[0]) {
+					m[x[0]] = x[1]
+				} else {
+					m[x[0]] = replaceSpecialCharacters(x[1])
+				}
 			}
 		} else {
 			if i-1 >= 0 {

From 1da742e315fd0c0afcf91a025b65965eb249ca9e Mon Sep 17 00:00:00 2001
From: Yifei Yuan <yyuan@confluent.io>
Date: Mon, 29 Jun 2026 19:57:03 -0400
Subject: [PATCH 02/10] add tests with json configs

---
 pkg/properties/properties_test.go | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/pkg/properties/properties_test.go b/pkg/properties/properties_test.go
index e25f795dfe..c259f7d718 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,15 @@ func TestCreateKeyValuePairsKeysWithDotsAndSorts(t *testing.T) {
 	m["connection.mode"] = "OUTBOUND"
 	require.Equal(t, "\"connection.mode\"=\"OUTBOUND\"\n\"link.mode\"=\"BIDIRECTIONAL\"\n", CreateKeyValuePairs(m))
 }
+
+func TestGetMap_FileWithRawValueKeyPreservesJSON(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "topic.properties")
+	// confluent.value.association is stored verbatim, so its JSON value must be preserved.
+	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 := GetMap([]string{path}, "confluent.value.association")
+	require.NoError(t, err)
+	require.Equal(t, json, m["confluent.value.association"])
+}

From c5f2590cbe69cb9f99883857e65179ef70c8cd89 Mon Sep 17 00:00:00 2001
From: Yifei Yuan <yyuan@confluent.io>
Date: Wed, 8 Jul 2026 16:41:32 -0400
Subject: [PATCH 03/10] Update integration tests and golden files

---
 test/fixtures/input/kafka/topic/topic-config-json.properties | 1 +
 test/fixtures/input/kafka/topic/topic-config.properties      | 2 ++
 2 files changed, 3 insertions(+)
 create mode 100644 test/fixtures/input/kafka/topic/topic-config-json.properties
 create mode 100644 test/fixtures/input/kafka/topic/topic-config.properties

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..f3f26bbfc1
--- /dev/null
+++ b/test/fixtures/input/kafka/topic/topic-config-json.properties
@@ -0,0 +1 @@
+confluent.key.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

From 4e238e7e63e79bd5bbbb5442dbbdc9a19b31ce83 Mon Sep 17 00:00:00 2001
From: Yifei Yuan <yyuan@confluent.io>
Date: Wed, 8 Jul 2026 16:46:34 -0400
Subject: [PATCH 04/10] Update golden file and kafka integration tests

---
 test/fixtures/output/kafka/topic/create-help.golden | 2 +-
 test/fixtures/output/kafka/topic/create.golden      | 3 +--
 test/kafka_test.go                                  | 2 ++
 test/test-server/kafka_rest_router.go               | 4 +++-
 4 files changed, 7 insertions(+), 4 deletions(-)

diff --git a/test/fixtures/output/kafka/topic/create-help.golden b/test/fixtures/output/kafka/topic/create-help.golden
index b7b983381b..8efedad2d9 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 strings          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..43281cad69 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 strings          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.
@@ -21,4 +21,3 @@ Global Flags:
   -h, --help            Show help for this command.
       --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
   -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
-
diff --git a/test/kafka_test.go b/test/kafka_test.go
index 3e146e84ae..990bb7c265 100644
--- a/test/kafka_test.go
+++ b/test/kafka_test.go
@@ -152,6 +152,8 @@ 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 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},
 
diff --git a/test/test-server/kafka_rest_router.go b/test/test-server/kafka_rest_router.go
index 06e2c204f2..7932ae333c 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"
@@ -216,8 +217,9 @@ func handleKafkaRestTopics(t *testing.T) http.HandlerFunc {
 				return
 			}
 			// check configs
+			allowedConfigNames := []string{"retention.ms", "compression.type", "confluent.key.association", "confluent.value.association"}
 			for _, config := range requestData.Configs {
-				if config.Name != "retention.ms" && config.Name != "compression.type" {
+				if !slices.Contains(allowedConfigNames, 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" {

From 55762b297864a50d9b931ed826638c9c6741df62 Mon Sep 17 00:00:00 2001
From: Yifei Yuan <yyuan@confluent.io>
Date: Tue, 14 Jul 2026 15:04:37 -0400
Subject: [PATCH 05/10] add integration tests for kafka update

---
 .../kafka/topic/topic-config-json.properties  |  3 +-
 .../kafka/topic/update-json-config.golden     |  6 +++
 test/kafka_test.go                            |  2 +
 test/test-server/kafka_rest_router.go         | 41 +++++++++++++++++--
 4 files changed, 48 insertions(+), 4 deletions(-)
 create mode 100644 test/fixtures/output/kafka/topic/update-json-config.golden

diff --git a/test/fixtures/input/kafka/topic/topic-config-json.properties b/test/fixtures/input/kafka/topic/topic-config-json.properties
index f3f26bbfc1..0a8bb7e825 100644
--- a/test/fixtures/input/kafka/topic/topic-config-json.properties
+++ b/test/fixtures/input/kafka/topic/topic-config-json.properties
@@ -1 +1,2 @@
-confluent.key.association={"schema":"{\"type\":\"record\",\"name\":\"TestRecord\",\"doc\":\"Basic test.\\na=b.\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"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\"]}]}"}
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..faf5557b24
--- /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 990bb7c265..29633f6569 100644
--- a/test/kafka_test.go
+++ b/test/kafka_test.go
@@ -180,6 +180,8 @@ 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"},
 	}
 
 	if runtime.GOOS != "windows" {
diff --git a/test/test-server/kafka_rest_router.go b/test/test-server/kafka_rest_router.go
index 7932ae333c..eb139463a0 100644
--- a/test/test-server/kafka_rest_router.go
+++ b/test/test-server/kafka_rest_router.go
@@ -28,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,9 +235,8 @@ func handleKafkaRestTopics(t *testing.T) http.HandlerFunc {
 				return
 			}
 			// check configs
-			allowedConfigNames := []string{"retention.ms", "compression.type", "confluent.key.association", "confluent.value.association"}
 			for _, config := range requestData.Configs {
-				if !slices.Contains(allowedConfigNames, config.Name) {
+				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" {
@@ -230,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
 			}
@@ -305,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)
@@ -516,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" {
@@ -527,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
 				}

From 580720fc2b132736116ecd3ead637e1ca1e36760 Mon Sep 17 00:00:00 2001
From: Yifei Yuan <yyuan@confluent.io>
Date: Tue, 14 Jul 2026 15:05:56 -0400
Subject: [PATCH 06/10] update golden

---
 .../fixtures/output/kafka/topic/update-json-config.golden | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/test/fixtures/output/kafka/topic/update-json-config.golden b/test/fixtures/output/kafka/topic/update-json-config.golden
index faf5557b24..40f7030907 100644
--- a/test/fixtures/output/kafka/topic/update-json-config.golden
+++ b/test/fixtures/output/kafka/topic/update-json-config.golden
@@ -1,6 +1,6 @@
 Updated the following configuration values for topic "topic-exist-rest":
-             Name             |                                        Value                                        | Read-Only
+             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\"]}]}"} |
+  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\"]}]}"} |            

From a6f7a4b145bfa66a28e44114f21081398c6c67be Mon Sep 17 00:00:00 2001
From: Yifei Yuan <yyuan@confluent.io>
Date: Thu, 16 Jul 2026 01:32:49 -0400
Subject: [PATCH 07/10] update golden

---
 test/fixtures/output/kafka/topic/create.golden | 1 +
 1 file changed, 1 insertion(+)

diff --git a/test/fixtures/output/kafka/topic/create.golden b/test/fixtures/output/kafka/topic/create.golden
index 43281cad69..83f9045d89 100644
--- a/test/fixtures/output/kafka/topic/create.golden
+++ b/test/fixtures/output/kafka/topic/create.golden
@@ -21,3 +21,4 @@ Global Flags:
   -h, --help            Show help for this command.
       --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
   -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
+

From a631baeef3ade266f8f672e6eb2a3f45f30839c8 Mon Sep 17 00:00:00 2001
From: Yifei Yuan <yyuan@confluent.io>
Date: Thu, 23 Jul 2026 18:40:29 -0400
Subject: [PATCH 08/10] Add new TopicConfigFlag parser for inline json-valued
 configs

---
 pkg/cmd/flags.go                  |  7 +++
 pkg/properties/properties.go      | 92 +++++++++++++++++++++++++++----
 pkg/properties/properties_test.go | 67 +++++++++++++++++++++-
 3 files changed, 153 insertions(+), 13 deletions(-)

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 b23f601121..baa42a64d5 100644
--- a/pkg/properties/properties.go
+++ b/pkg/properties/properties.go
@@ -2,6 +2,7 @@ package properties
 
 import (
 	"bytes"
+	"encoding/json"
 	"fmt"
 	"os"
 	"slices"
@@ -12,13 +13,12 @@ import (
 )
 
 // GetMap reads newline-separated configuration files or comma-separated lists of key=value pairs, and supports configuration values containing commas.
-// Values whose keys are listed in rawValueKeys are stored verbatim, skipping special characters un-escaping, to preserve values such as JSON strings unchanged.
-func GetMap(config []string, rawValueKeys ...string) (map[string]string, error) {
+func GetMap(config []string) (map[string]string, error) {
 	if len(config) == 1 && utils.FileExists(config[0]) {
-		return fileToMap(config[0], rawValueKeys...)
+		return fileToMap(config[0])
 	}
 
-	return ConfigFlagToMap(config, rawValueKeys...)
+	return ConfigFlagToMap(config)
 }
 
 // fileToMap reads key=value pairs from a properties file, ignoring comments and empty lines.
@@ -69,18 +69,14 @@ func ParseLines(content string) []string {
 }
 
 // ConfigFlagToMap reads key=values pairs from the --config flag and supports configuration values containing commas.
-func ConfigFlagToMap(configs []string, rawValueKeys ...string) (map[string]string, error) {
+func ConfigFlagToMap(configs []string) (map[string]string, error) {
 	m := make(map[string]string)
 
 	for i := len(configs) - 1; i >= 0; i-- {
 		if strings.Contains(configs[i], "=") {
 			x := strings.SplitN(configs[i], "=", 2)
 			if _, ok := m[x[0]]; !ok {
-				if slices.Contains(rawValueKeys, x[0]) {
-					m[x[0]] = x[1]
-				} else {
-					m[x[0]] = replaceSpecialCharacters(x[1])
-				}
+				m[x[0]] = replaceSpecialCharacters(x[1])
 			}
 		} else {
 			if i-1 >= 0 {
@@ -94,6 +90,82 @@ func ConfigFlagToMap(configs []string, rawValueKeys ...string) (map[string]strin
 	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 c259f7d718..a2548f962c 100644
--- a/pkg/properties/properties_test.go
+++ b/pkg/properties/properties_test.go
@@ -127,14 +127,75 @@ func TestCreateKeyValuePairsKeysWithDotsAndSorts(t *testing.T) {
 	require.Equal(t, "\"connection.mode\"=\"OUTBOUND\"\n\"link.mode\"=\"BIDIRECTIONAL\"\n", CreateKeyValuePairs(m))
 }
 
-func TestGetMap_FileWithRawValueKeyPreservesJSON(t *testing.T) {
+// 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")
-	// confluent.value.association is stored verbatim, so its JSON value must be preserved.
+	// 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 := GetMap([]string{path}, "confluent.value.association")
+	m, err := GetMapFromArray([]string{path}, "confluent.value.association")
 	require.NoError(t, err)
 	require.Equal(t, json, m["confluent.value.association"])
 }

From 47af904a02c1849e46462a2f36579de60743c5c5 Mon Sep 17 00:00:00 2001
From: Yifei Yuan <yyuan@confluent.io>
Date: Thu, 23 Jul 2026 18:46:16 -0400
Subject: [PATCH 09/10] read inline JSON-valued configs on kafka topic
 create/update

---
 internal/kafka/command_topic_create.go | 13 +++++++------
 internal/kafka/command_topic_update.go |  6 +++---
 2 files changed, 10 insertions(+), 9 deletions(-)

diff --git a/internal/kafka/command_topic_create.go b/internal/kafka/command_topic_create.go
index aa31422883..93544eb324 100644
--- a/internal/kafka/command_topic_create.go
+++ b/internal/kafka/command_topic_create.go
@@ -19,9 +19,10 @@ import (
 	"github.com/confluentinc/cli/v4/pkg/utils"
 )
 
-// confluent.*.association values are JSON strings. They must be stored verbatim,
-// so keys are passed as rawValueKeys to skip special-character un-escaping.
-var rawValueTopicConfigs = []string{"confluent.key.association", "confluent.value.association"}
+// 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{
@@ -39,7 +40,7 @@ func (c *command) newCreateCommand() *cobra.Command {
 	}
 
 	cmd.Flags().Uint32("partitions", 0, "Number of topic partitions.")
-	pcmd.AddConfigFlag(cmd)
+	pcmd.AddTopicConfigFlag(cmd)
 	pcmd.AddEndpointFlag(cmd, c.AuthenticatedCLICommand)
 	pcmd.AddDryRunFlag(cmd)
 	cmd.Flags().Bool("if-not-exists", false, "Exit gracefully if topic already exists.")
@@ -58,11 +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.GetMap(configs, rawValueTopicConfigs...)
+	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 24a007d5d5..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, rawValueTopicConfigs...)
+	configMap, err := properties.GetMapFromArray(configs, associationConfigs...)
 	if err != nil {
 		return err
 	}

From 10311b0ad9569992d7441a4c0ea47d404ddc489f Mon Sep 17 00:00:00 2001
From: Yifei Yuan <yyuan@confluent.io>
Date: Thu, 23 Jul 2026 19:15:12 -0400
Subject: [PATCH 10/10] Add integration tests and golden

---
 test/fixtures/output/kafka/topic/create-help.golden | 2 +-
 test/fixtures/output/kafka/topic/create.golden      | 2 +-
 test/fixtures/output/kafka/topic/update-help.golden | 2 +-
 test/kafka_test.go                                  | 4 ++++
 4 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/test/fixtures/output/kafka/topic/create-help.golden b/test/fixtures/output/kafka/topic/create-help.golden
index 8efedad2d9..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 "key=value" pairs, or path to a configuration file containing a newline-separated list of "key=value" pairs.
+      --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 83f9045d89..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 "key=value" pairs, or path to a configuration file containing a newline-separated list of "key=value" pairs.
+      --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/kafka_test.go b/test/kafka_test.go
index 29633f6569..521c364bd8 100644
--- a/test/kafka_test.go
+++ b/test/kafka_test.go
@@ -154,6 +154,9 @@ func (s *CLITestSuite) TestKafka() {
 		{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},
 
@@ -182,6 +185,7 @@ func (s *CLITestSuite) TestKafka() {
 		{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" {