Merge pull request #189 from checkr/zz/add-kafka-version

Add kafka version config
openflagr · Nov 13, 2018 · f28b839 · f28b839
2 parents 4b64006 + 8e84a94
commit f28b839
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 0 deletions.
diff --git a/pkg/config/env.go b/pkg/config/env.go
@@ -67,6 +67,7 @@ var Config = struct {
 	RecorderType string `env:"FLAGR_RECORDER_TYPE" envDefault:"kafka"`
 
 	// Kafka related configurations for data records logging (Flagr Metrics)
+	RecorderKafkaVersion        string        `env:"FLAGR_RECORDER_KAFKA_VERSION" envDefault:"0.8.2.0"`
 	RecorderKafkaBrokers        string        `env:"FLAGR_RECORDER_KAFKA_BROKERS" envDefault:":9092"`
 	RecorderKafkaCertFile       string        `env:"FLAGR_RECORDER_KAFKA_CERTFILE" envDefault:""`
 	RecorderKafkaKeyFile        string        `env:"FLAGR_RECORDER_KAFKA_KEYFILE" envDefault:""`

diff --git a/pkg/handler/data_recorder_kafka.go b/pkg/handler/data_recorder_kafka.go
@@ -23,6 +23,14 @@ var (
 	saramaNewAsyncProducer = sarama.NewAsyncProducer
 )
 
+func mustParseKafkaVersion() sarama.KafkaVersion {
+	v, err := sarama.ParseKafkaVersion(config.Config.RecorderKafkaVersion)
+	if err != nil {
+		panic(err)
+	}
+	return v
+}
+
 // NewKafkaRecorder creates a new Kafka recorder
 var NewKafkaRecorder = func() DataRecorder {
 	cfg := sarama.NewConfig()
@@ -39,6 +47,7 @@ var NewKafkaRecorder = func() DataRecorder {
 	cfg.Producer.RequiredAcks = sarama.WaitForLocal
 	cfg.Producer.Retry.Max = config.Config.RecorderKafkaRetryMax
 	cfg.Producer.Flush.Frequency = config.Config.RecorderKafkaFlushFrequency
+	cfg.Version = mustParseKafkaVersion()
 
 	brokerList := strings.Split(config.Config.RecorderKafkaBrokers, ",")
 	producer, err := saramaNewAsyncProducer(brokerList, cfg)