diff --git a/charts/fission-all/values.yaml b/charts/fission-all/values.yaml index cff3d6b2..c476152c 100644 --- a/charts/fission-all/values.yaml +++ b/charts/fission-all/values.yaml @@ -74,7 +74,10 @@ kafka: enabled: false brokers: 'broker.kafka:9092' ## 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 # version: "0.11.0.0" diff --git a/mqtrigger/messageQueue/kafka.go b/mqtrigger/messageQueue/kafka.go index 3f657265..204498ed 100644 --- a/mqtrigger/messageQueue/kafka.go +++ b/mqtrigger/messageQueue/kafka.go @@ -151,8 +151,12 @@ func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *crd.Me // Set the headers came from Kafka record // Using Header.Add() as msg.Headers may have keys with more than one value - for _, h := range msg.Headers { - req.Header.Add(string(h.Key), string(h.Value)) + if kafka.version.IsAtLeast(sarama.V0_11_0_0) { + 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 {