read record header as metadata if available in kafka
@@ -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, "")
