fix(kafka): fix dynamic prop problem

Signed-off-by: Jiyong Huang <huangjy@emqx.io>
ngjaying · Nov 21, 2024 · 35e6c51 · 35e6c51
1 parent 3875ce3
commit 35e6c51
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/extensions/impl/kafka/sink.go b/extensions/impl/kafka/sink.go
@@ -208,7 +208,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
 			}
@@ -252,7 +252,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
 				}
@@ -267,7 +267,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
 			}