Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[Kafka MQT] Add warning about Kafka version (#1083)

Browse Source 
For Kafka version < 0.11.0.0 log a warning about headers not being passed in the HTTP request
This commit is contained in:
Bhavin Gandhi
2019-02-26 16:35:17 +05:30
committed by Vishal
parent d65a0cb9dc
commit 3358fca753
2 changed files with 10 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
charts/fission-all/values.yaml
+4 -1
View File
@@ -74,7 +74,10 @@ kafka:
enabled: false enabled: false
brokers: 'broker.kafka:9092' brokers: 'broker.kafka:9092'
## version of Kafka broker ## version of Kafka broker
## Must be a string in the format "major.minor.veryMinor.patch" ## For 0.x it must be a string in the format
## "major.minor.veryMinor.patch" example: 0.8.2.0
## For 1.x it must be a string in the format
## "major.major.veryMinor" example: 2.0.1
## Should be >= 0.11.0.0 to enable Kafka record headers support ## Should be >= 0.11.0.0 to enable Kafka record headers support
# version: "0.11.0.0" # version: "0.11.0.0"
mqtrigger/messageQueue/kafka.go
+6 -2
View File
@@ -151,8 +151,12 @@ func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *crd.Me
// Set the headers came from Kafka record // Set the headers came from Kafka record
// Using Header.Add() as msg.Headers may have keys with more than one value // Using Header.Add() as msg.Headers may have keys with more than one value
for _, h := range msg.Headers { if kafka.version.IsAtLeast(sarama.V0_11_0_0) {
req.Header.Add(string(h.Key), string(h.Value)) for _, h := range msg.Headers {
req.Header.Add(string(h.Key), string(h.Value))
}
} else {
log.Warningf("Headers are not supported by Kafka version %q, needs v0.11+: no record headers to add in HTTP request", kafka.version)
} }
for k, v := range fissionHeaders { for k, v := range fissionHeaders {