kafkamdm: make kafka version configurable

cfg/cfg.go

@@ -105,6 +105,7 @@ type Route struct {
 	Topic         string // also used by Google PubSub
 	Codec         string // also used by Google PubSub
 	PartitionBy   string
+	KafkaVersion  string
 	TLSEnabled    bool
 	TLSSkipVerify bool
 	TLSClientCert string

docs/config.md

@@ -218,6 +218,7 @@ topic          |     Y     |  string     | N/A     |
 codec          |     Y     |  string     | N/A     | which compression to use. possible values: none, gzip, snappy
 partitionBy    |     Y     |  string     | N/A     | which fields to shard by. possible values are: byOrg, bySeries, bySeriesWithTags
 schemasFile    |     Y     |  string     | N/A     |
+kafkaVersion   |     N     |  string     | ""      | Kafka version in semver format. All brokers must be this version or newer. (fallback to oldest stable version 1.0.0)
 prefix         |     N     |  string     | ""      |
 notPrefix      |     N     |  string     | ""      |
 sub            |     N     |  string     | ""      |
@@ -248,6 +249,7 @@ brokers = ['kafka:9092']
 topic = 'mdm'
 codec = 'snappy'
 partitionBy = 'bySeriesWithTags'
+kafkaVersion = '2.0.0'
 schemasFile = 'conf/storage-schemas.conf'
 tlsEnabled = true
 tlsSkipVerify  = false

imperatives/imperatives.go

@@ -68,6 +68,7 @@ const (
 	optSASLEnabled
 	optSASLUsername
 	optSASLPassword
+	optKafkaVersion
 	optUnspoolSleep
 	optPickle
 	optSpool
@@ -131,6 +132,7 @@ var tokens = []toki.Def{
 	{Token: optSASLEnabled, Pattern: "saslEnabled="},
 	{Token: optSASLUsername, Pattern: "saslUsername="},
 	{Token: optSASLPassword, Pattern: "saslPassword="},
+	{Token: optKafkaVersion, Pattern: "kafkaVersion="},
 	{Token: optUnspoolSleep, Pattern: "unspoolsleep="},
 	{Token: optPickle, Pattern: "pickle="},
 	{Token: optSpool, Pattern: "spool="},
@@ -696,6 +698,7 @@ func readAddRouteKafkaMdm(s *toki.Scanner, table Table) error {
 	var blocking = false
 	var tlsEnabled, tlsSkipVerify bool
 	var tlsClientCert, tlsClientKey string
+	var kafkaVersion string
 	var saslEnabled bool
 	var saslUsername, saslPassword string
 
@@ -806,12 +809,18 @@ func readAddRouteKafkaMdm(s *toki.Scanner, table Table) error {
 				return errFmtAddRouteKafkaMdm
 			}
 			saslPassword = string(t.Value)
+		case optKafkaVersion:
+			t = s.Next()
+			if t.Token != word {
+				return errFmtAddRouteKafkaMdm
+			}
+			kafkaVersion = string(t.Value)
 		default:
 			return fmt.Errorf("unexpected token %d %q", t.Token, t.Value)
 		}
 	}
 
-	route, err := route.NewKafkaMdm(key, matcher, topic, codec, schemasFile, partitionBy, brokers, bufSize, orgId, flushMaxNum, flushMaxWait, timeout, blocking, tlsEnabled, tlsSkipVerify, tlsClientCert, tlsClientKey, saslEnabled, saslUsername, saslPassword)
+	route, err := route.NewKafkaMdm(key, matcher, topic, codec, schemasFile, partitionBy, kafkaVersion, brokers, bufSize, orgId, flushMaxNum, flushMaxWait, timeout, blocking, tlsEnabled, tlsSkipVerify, tlsClientCert, tlsClientKey, saslEnabled, saslUsername, saslPassword)
 	if err != nil {
 		return err
 	}

route/kafkamdm.go

@@ -32,8 +32,8 @@ type KafkaMdm struct {
 	schemas       persister.WhisperSchemas
 	blocking      bool
 	dispatch      func(chan []byte, []byte, metrics.Gauge, metrics.Counter)
-
-	orgId int // organisation to publish data under
+	kafkaVersion  sarama.KafkaVersion
+	orgId         int // organisation to publish data under
 
 	bufSize      int // amount of messages we can buffer up. each message is about 100B. so 1e7 is about 1GB.
 	flushMaxNum  int
@@ -52,7 +52,7 @@ type KafkaMdm struct {
 
 // NewKafkaMdm creates a special route that writes to a grafana.net datastore
 // We will automatically run the route and the destination
-func NewKafkaMdm(key string, matcher matcher.Matcher, topic, codec, schemasFile, partitionBy string, brokers []string, bufSize, orgId, flushMaxNum, flushMaxWait, timeout int, blocking bool, tlsEnabled, tlsSkipVerify bool, tlsClientCert, tlsClientKey string, saslEnabled bool, saslUsername, saslPassword string) (Route, error) {
+func NewKafkaMdm(key string, matcher matcher.Matcher, topic, codec, schemasFile, partitionBy, kafkaVersion string, brokers []string, bufSize, orgId, flushMaxNum, flushMaxWait, timeout int, blocking bool, tlsEnabled, tlsSkipVerify bool, tlsClientCert, tlsClientKey string, saslEnabled bool, saslUsername, saslPassword string) (Route, error) {
 	schemas, err := getSchemas(schemasFile)
 	if err != nil {
 		return nil, err
@@ -69,10 +69,9 @@ func NewKafkaMdm(key string, matcher matcher.Matcher, topic, codec, schemasFile,
 		blocking:  blocking,
 		orgId:     orgId,
 
-		bufSize:      bufSize,
-		flushMaxNum:  flushMaxNum,
-		flushMaxWait: time.Duration(flushMaxWait) * time.Millisecond,
-
+		bufSize:           bufSize,
+		flushMaxNum:       flushMaxNum,
+		flushMaxWait:      time.Duration(flushMaxWait) * time.Millisecond,
 		numErrFlush:       stats.Counter("dest=" + cleanAddr + ".unit=Err.type=flush"),
 		numOut:            stats.Counter("dest=" + cleanAddr + ".unit=Metric.direction=out"),
 		durationTickFlush: stats.Timer("dest=" + cleanAddr + ".what=durationFlush.type=ticker"),
@@ -96,11 +95,19 @@ func NewKafkaMdm(key string, matcher matcher.Matcher, topic, codec, schemasFile,
 		log.Fatalf("kafkaMdm %q: failed to initialize partitioner. %s", r.key, err)
 	}
 
+	kfkVersion, err := sarama.ParseKafkaVersion(kafkaVersion)
+	if err != nil {
+		log.Warnf("kafkaMdm %q: failed to parse kafka version fallback to default version. %s", r.key, err)
+		//ParseKafkaVersion() will return an error and sarama.DefaultVersion
+		//this means that we have the same behavior that we had before this was configurable.
+	}
+	r.kafkaVersion = kfkVersion
+
 	// We are looking for strong consistency semantics.
 	// Because we don't change the flush settings, sarama will try to produce messages
 	// as fast as possible to keep latency low.
 	config := sarama.NewConfig()
-
+	config.Version = r.kafkaVersion
 	if tlsEnabled {
 		tlsConfig, err := tls.NewConfig(tlsClientCert, tlsClientKey)
 		if err != nil {

table/table.go

@@ -845,7 +845,7 @@ func (table *Table) InitRoutes(config cfg.Config, meta toml.MetaData) error {
 				orgId = routeConfig.OrgId
 			}
 
-			route, err := route.NewKafkaMdm(routeConfig.Key, matcher, routeConfig.Topic, routeConfig.Codec, routeConfig.SchemasFile, routeConfig.PartitionBy, routeConfig.Brokers, bufSize, orgId, flushMaxNum, flushMaxWait, timeout, routeConfig.Blocking, routeConfig.TLSEnabled, routeConfig.TLSSkipVerify, routeConfig.TLSClientCert, routeConfig.TLSClientKey, routeConfig.SASLEnabled, routeConfig.SASLUsername, routeConfig.SASLPassword)
+			route, err := route.NewKafkaMdm(routeConfig.Key, matcher, routeConfig.Topic, routeConfig.Codec, routeConfig.SchemasFile, routeConfig.PartitionBy, routeConfig.KafkaVersion, routeConfig.Brokers, bufSize, orgId, flushMaxNum, flushMaxWait, timeout, routeConfig.Blocking, routeConfig.TLSEnabled, routeConfig.TLSSkipVerify, routeConfig.TLSClientCert, routeConfig.TLSClientKey, routeConfig.SASLEnabled, routeConfig.SASLUsername, routeConfig.SASLPassword)
 			if err != nil {
 				log.Error(err.Error())
 				return fmt.Errorf("error adding route '%s'", routeConfig.Key)