Repository: lf-edge/ekuiper (go)

Issue:
Kafka sink -  key not filled in correctly in Version 2.0
**Environment**:

- eKuiper version:  2.0-full
- Hardware configuration (e.g. `lscpu`):
![image](https://github.com/user-attachments/assets/23cdf24d-9ccd-455e-9121-71288cddee61)
OS (e.g. cat /etc/os-release): Debian GNU/Linux 11 (bullseye)"
Others: deploymen as docker container, image version: docker pull lfedge/ekuiper:2.0-full

The key attribute in the Kafka sink configuration is not correctly populated in eKuiper version 1.14.6+ when using a pattern-based key assignment. The issue was not present in version 1.14.6, where the key was correctly derived from the events generated by the SQL query.

Steps to Reproduce:

1. Kafka Sink Configuration: The following Kafka sink configuration is used:
```
{
    "kafka": {
        "batchSize": 1,
        "brokers": "192.168.xxx.xx:9092",
        "bufferLength": 1024,
        "enableCache": false,
        "format": "json",
        "insecureSkipVerify": false,
        "key": "{{.key}}",
        "maxAttempts": 1,
        "omitIfEmpty": false,
        "requiredACKs": 1,
        "resourceId": "kafka",
        "runAsync": false,
        "saslAuthType": "none",
        "sendSingle": true,
        "topic": "aggregation.room"
    }
}

```
2. Event Stream Data: The SQL query produces events with the following structure:
```
{
    "humidity": 48.2,
    "key": "TempHumPres",
    "topic": "TempHumPres/get/sensor/2"
}

{
    "key": "TempHumPres",
    "temperature": 22.4,
    "topic": "TempHumPres/get/sensor/1"
}

{
    "key": "TempHumPres",
    "pressure": 1005.7,
    "topic": "TempHumPres/get/sensor/3"
}

```
 3. Kafka Topic Observation: The Kafka topic receives messages with  key: {{.key}} instead of the expected value derived from the key attribute in the events. Expected Key: TempHumPres
 
![image](https://github.com/user-attachments/assets/1788777e-c5bf-4f40-a738-8cadec1a4274)

4. Previous Behavior: In version 1.14.6, the same configuration correctly populated the Kafka key based on the key attribute in the event.

![image](https://github.com/user-attachments/assets/4442fed3-e664-4401-8337-beb8d425018b)



Pull request description:
fix(kafka): fix dynamic prop problem
Closes #3389

Patch:
diff --git a/extensions/impl/kafka/sink.go b/extensions/impl/kafka/sink.go
index fdba9ddc..defbb978 100644
--- 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
 			}


Test patch:
diff --git a/extensions/impl/kafka/sink_test.go b/extensions/impl/kafka/sink_test.go
index f618b620..ee358fb8 100644
--- 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)