kafka_consumer: expose partition assignment strategy configuration (#6687)

Gopkg.toml

@@ -148,8 +148,7 @@
 
 [[constraint]]
   name = "github.com/Shopify/sarama"
-  revision = "b12709e6ca29240128c89fe0b30b6a76be42b457"
-  source = "https://github.com/influxdata/sarama.git"
+  version = "1.24.0"
 
 [[constraint]]
   name = "github.com/soniah/gosnmp"

etc/telegraf.conf

@@ -5116,6 +5116,9 @@
 #   ## Initial offset position; one of "oldest" or "newest".
 #   # offset = "oldest"
 #
+#   ## Consumer group partition assignment strategy; one of "range", "roundrobin" or "sticky".
+#   # balance_strategy = "range"
+#
 #   ## Maximum length of a message to consume, in bytes (default 0/unlimited);
 #   ## larger messages are dropped
 #   max_message_len = 1000000

plugins/inputs/kafka_consumer/README.md

@@ -44,6 +44,9 @@ and use the old zookeeper connection method.
   ## Initial offset position; one of "oldest" or "newest".
   # offset = "oldest"
 
+  ## Consumer group partition assignment strategy; one of "range", "roundrobin" or "sticky".
+  # balance_strategy = "range"
+
   ## Maximum length of a message to consume, in bytes (default 0/unlimited);
   ## larger messages are dropped
   max_message_len = 1000000

plugins/inputs/kafka_consumer/kafka_consumer.go

@@ -49,6 +49,9 @@ const sampleConfig = `
   ## Initial offset position; one of "oldest" or "newest".
   # offset = "oldest"
 
+  ## Consumer group partition assignment strategy; one of "range", "roundrobin" or "sticky".
+  # balance_strategy = "range"
+
   ## Maximum length of a message to consume, in bytes (default 0/unlimited);
   ## larger messages are dropped
   max_message_len = 1000000
@@ -86,6 +89,7 @@ type KafkaConsumer struct {
 	MaxMessageLen          int      `toml:"max_message_len"`
 	MaxUndeliveredMessages int      `toml:"max_undelivered_messages"`
 	Offset                 string   `toml:"offset"`
+	BalanceStrategy        string   `toml:"balance_strategy"`
 	Topics                 []string `toml:"topics"`
 	TopicTag               string   `toml:"topic_tag"`
 	Version                string   `toml:"version"`
@@ -185,6 +189,17 @@ func (k *KafkaConsumer) Init() error {
 		return fmt.Errorf("invalid offset %q", k.Offset)
 	}
 
+	switch strings.ToLower(k.BalanceStrategy) {
+	case "range", "":
+		config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
+	case "roundrobin":
+		config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
+	case "sticky":
+		config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategySticky
+	default:
+		return fmt.Errorf("invalid balance strategy %q", k.BalanceStrategy)
+	}
+
 	if k.ConsumerCreator == nil {
 		k.ConsumerCreator = &SaramaCreator{}
 	}