✨ adding header to consumer group as well

Signed-off-by: Clement Montois <[email protected]>
cmontois · Apr 11, 2023 · f7fd76a · f7fd76a
1 parent 92f5612
commit f7fd76a
Showing 1 changed file with 9 additions and 0 deletions.
diff --git a/eventsources/sources/kafka/start.go b/eventsources/sources/kafka/start.go
@@ -399,6 +399,15 @@ func (consumer *Consumer) processOne(session sarama.ConsumerGroupSession, messag
 		Timestamp: message.Timestamp.String(),
 		Metadata:  consumer.kafkaEventSource.Metadata,
 	}
+
+	headers := make(map[string]string)
+
+	for _, recordHeader := range message.Headers {
+		headers[string(recordHeader.Key)] = string(recordHeader.Value)
+	}
+
+	eventData.Headers = headers
+
 	if consumer.kafkaEventSource.JSONBody {
 		eventData.Body = (*json.RawMessage)(&message.Value)
 	} else {