read record header as metadata if available in kafka

Signed-off-by: Mukundan Sundararajan <65565396+mukundansundar@users.noreply.github.com>

internal/component/kafka/consumer.go

@@ -67,6 +67,13 @@ func (consumer *consumer) doCallback(session sarama.ConsumerGroupSession, messag
 		Topic: message.Topic,
 		Data:  message.Value,
 	}
+	// This is true only when headers are set (Kafka > 0.11)
+	if message.Headers != nil && len(message.Headers) > 0 {
+		event.Metadata = make(map[string]string, len(message.Headers))
+		for _, header := range message.Headers {
+			event.Metadata[string(header.Key)] = string(header.Value)
+		}
+	}
 	err = handler(session.Context(), &event)
 	if err == nil {
 		session.MarkMessage(message, "")