Kafka PubSub: Use metadata as message headers. (#1409)

* Kafka PubSub: Setting metadata as message headers. Signed-off-by: Phil Kedy <phil.kedy@gmail.com> * running go mod tidy on certification tests Signed-off-by: Phil Kedy <phil.kedy@gmail.com>
dapr · Jan 6, 2022 · 265fa9a · 265fa9a
1 parent bc2b5cc
commit 265fa9a
Showing 1 changed file with 12 additions and 2 deletions.
diff --git a/pubsub/kafka/kafka.go b/pubsub/kafka/kafka.go
@@ -231,8 +231,18 @@ func (k *Kafka) Publish(req *pubsub.PublishRequest) error {
 		Value: sarama.ByteEncoder(req.Data),
 	}
 
-	if val, ok := req.Metadata[key]; ok && val != "" {
-		msg.Key = sarama.StringEncoder(val)
+	for name, value := range req.Metadata {
+		if name == key {
+			msg.Key = sarama.StringEncoder(value)
+		} else {
+			if msg.Headers == nil {
+				msg.Headers = make([]sarama.RecordHeader, 0, len(req.Metadata))
+			}
+			msg.Headers = append(msg.Headers, sarama.RecordHeader{
+				Key:   []byte(name),
+				Value: []byte(value),
+			})
+		}
 	}
 
 	partition, offset, err := k.producer.SendMessage(msg)