From 368eaa54488672c3bc952a008ed39384cf7d6691 Mon Sep 17 00:00:00 2001
From: Eduardas Kazakas <eduardas.kazakas@vinted.com>
Date: Fri, 20 Dec 2024 12:24:54 +0200
Subject: [PATCH 1/2] fix(conf): improve type conversion

---
 kafka/config.go | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/kafka/config.go b/kafka/config.go
index d9174b49f..565b88529 100644
--- a/kafka/config.go
+++ b/kafka/config.go
@@ -57,7 +57,11 @@ func (m ConfigMap) SetKey(key string, value ConfigValue) error {
 		if !found {
 			m["default.topic.config"] = ConfigMap{}
 		}
-		m["default.topic.config"].(ConfigMap)[strings.TrimPrefix(key, "{topic}.")] = value
+		if cm, ok := m["default.topic.config"].(ConfigMap); !ok {
+			return newErrorFromString(ErrInvalidArg, "value type is not a ConfigMap")
+		} else {
+			cm[strings.TrimPrefix(key, "{topic}.")] = value
+		}
 	} else {
 		m[key] = value
 	}

From 0a60ed9dec35e4254f34b1c03e563c0c81ae193a Mon Sep 17 00:00:00 2001
From: Eduardas Kazakas <eduardas.kazakas@vinted.com>
Date: Fri, 20 Dec 2024 12:45:46 +0200
Subject: [PATCH 2/2] WIP

---
 kafka/config.go      |  7 ++++++-
 kafka/config_test.go | 29 +++++++++++++++++++++++++++++
 2 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/kafka/config.go b/kafka/config.go
index 565b88529..c843b07cb 100644
--- a/kafka/config.go
+++ b/kafka/config.go
@@ -232,7 +232,12 @@ func (m ConfigMap) get(key string, defval ConfigValue) (ConfigValue, error) {
 		if !found {
 			return defval, nil
 		}
-		return defconfCv.(ConfigMap).get(strings.TrimPrefix(key, "{topic}."), defval)
+
+		if cm, ok := defconfCv.(ConfigMap); !ok {
+			return nil, newErrorFromString(ErrInvalidArg, "value type is not a ConfigMap")
+		} else {
+			return cm.get(strings.TrimPrefix(key, "{topic}."), defval)
+		}
 	}
 
 	v, ok := m[key]
diff --git a/kafka/config_test.go b/kafka/config_test.go
index cdd2ea725..f8a303c66 100644
--- a/kafka/config_test.go
+++ b/kafka/config_test.go
@@ -123,6 +123,35 @@ func TestConfigMapAPIs(t *testing.T) {
 		t.Errorf("Expected nil for dummy value, got %v\n", v)
 	}
 
+	config2 := &ConfigMap{
+		"default.topic.config": "dummy",
+	}
+	_, err = config2.Get("{topic}.produce.offset.report", false)
+	if err == nil {
+		t.Errorf("Expected Get({topic}.produce.offset.report) to fail\n")
+	}
+	err = config2.SetKey("{topic}.produce.offset.report", true)
+	if err == nil {
+		t.Errorf("Expected SetKey({topic}.produce.offset.report) to fail\n")
+	}
+
+	config3 := &ConfigMap{
+		"default.topic.config": ConfigMap{},
+	}
+	err = config3.SetKey("{topic}.produce.offset.report", true)
+	if err != nil {
+		t.Errorf("Expected SetKey({topic}.produce.offset.report) to succeed: %s\n", err)
+	}
+	v, err = config3.Get("{topic}.produce.offset.report", false)
+	if err != nil {
+		t.Errorf("Expected Get({topic}.produce.offset.report) to succeed: %s\n", err)
+	}
+	if v == nil {
+		t.Errorf("Expected Get({topic}.produce.offset.report) to return non-nil value\n")
+	}
+	if v.(bool) != true {
+		t.Errorf("produce.offset.report mismatch: %v\n", v.(bool))
+	}
 }
 
 // Test that plugins will always be configured before their config options