fix(kafka): fix dynamic prop problem

Signed-off-by: Jiyong Huang <huangjy@emqx.io>
ngjaying · Nov 21, 2024 · 19705f0 · 19705f0
1 parent 56df9c5
commit 19705f0
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 5 deletions.
diff --git a/extensions/impl/kafka/sink.go b/extensions/impl/kafka/sink.go
@@ -212,7 +212,7 @@ func (k *KafkaSink) buildMsg(ctx api.StreamContext, item api.MessageTuple, decod
 	if len(k.kc.Key) > 0 {
 		newKey := k.kc.Key
 		if dp, ok := item.(api.HasDynamicProps); ok {
-			key, ok := dp.DynamicProps("key")
+			key, ok := dp.DynamicProps(k.kc.Key)
 			if ok {
 				newKey = key
 			}
@@ -256,7 +256,7 @@ func (k *KafkaSink) parseHeaders(ctx api.StreamContext, item api.MessageTuple) (
 			value := v
 			dp, ok := item.(api.HasDynamicProps)
 			if ok {
-				nv, ok := dp.DynamicProps(k)
+				nv, ok := dp.DynamicProps(v)
 				if ok {
 					value = nv
 				}
@@ -271,7 +271,7 @@ func (k *KafkaSink) parseHeaders(ctx api.StreamContext, item api.MessageTuple) (
 		raw := k.headerTemplate
 		dp, ok := item.(api.HasDynamicProps)
 		if ok {
-			nv, ok := dp.DynamicProps("headers")
+			nv, ok := dp.DynamicProps(k.headerTemplate)
 			if ok {
 				raw = nv
 			}

diff --git a/extensions/impl/kafka/sink_test.go b/extensions/impl/kafka/sink_test.go
@@ -118,7 +118,7 @@ func TestKafkaSinkBuildMsg(t *testing.T) {
 	d, _ := json.Marshal(item)
 	mockT := testx.MockTuple{
 		Map:      item,
-		Template: map[string]string{"a": "1", "key": "1"},
+		Template: map[string]string{"{{.a}}": "1"},
 	}
 	msg, err := ks.buildMsg(ctx, mockT, d)
 	require.NoError(t, err)
@@ -141,7 +141,7 @@ func TestKafkaSinkBuildMsg(t *testing.T) {
 	}))
 	mockT = testx.MockTuple{
 		Map:      item,
-		Template: map[string]string{"headers": "{\"a\":\"1\"}"},
+		Template: map[string]string{"{\"a\":\"{{.a}}\"}": "{\"a\":\"1\"}"},
 	}
 	msg, err = ks.buildMsg(ctx, mockT, d)
 	require.NoError(t, err)