dapr
/
components-contrib
mirror of https://github.com/dapr/components-contrib.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add back headers metadata pass in kafka subscribe (#2281) 

* Add back headers metadata pass in kafka subscribe

Signed-off-by: Deepanshu Agarwal <deepanshu.agarwal1984@gmail.com>

* Update internal/component/kafka/consumer.go

Co-authored-by: Alessandro (Ale) Segala <43508+ItalyPaleAle@users.noreply.github.com>
Signed-off-by: Bernd Verst <4535280+berndverst@users.noreply.github.com>

Signed-off-by: Deepanshu Agarwal <deepanshu.agarwal1984@gmail.com>
Signed-off-by: Bernd Verst <4535280+berndverst@users.noreply.github.com>
Co-authored-by: Mukundan Sundararajan <65565396+mukundansundar@users.noreply.github.com>
Co-authored-by: Bernd Verst <4535280+berndverst@users.noreply.github.com>
Co-authored-by: Alessandro (Ale) Segala <43508+ItalyPaleAle@users.noreply.github.com>
Co-authored-by: Dapr Bot <56698301+dapr-bot@users.noreply.github.com>
This commit is contained in:
Deepanshu Agarwal 2022-12-21 01:13:52 +05:30 committed by GitHub
parent c156d4ba04
commit 2582f154bd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 7 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
internal/component/kafka/consumer.go
View File

@ -188,7 +188,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 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 = handlerConfig.Handler(session.Context(), &event)
if err == nil {
session.MarkMessage(message, "")