CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(inputs.kafka): switch to sarama's new consumer group rebalance strategy setting (#12064)

This commit is contained in:
reimda 2022-10-21 03:08:45 -06:00 committed by GitHub
parent 89cdfa35f7
commit a6352d9794
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 3 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
plugins/inputs/kafka_consumer/kafka_consumer.go
View File

@ -112,11 +112,11 @@ func (k *KafkaConsumer) Init() error {
switch strings.ToLower(k.BalanceStrategy) { switch strings.ToLower(k.BalanceStrategy) {
case "range", "": case "range", "":
cfg.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange cfg.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{sarama.BalanceStrategyRange}
case "roundrobin": case "roundrobin":
cfg.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin cfg.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{sarama.BalanceStrategyRoundRobin}
case "sticky": case "sticky":
cfg.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategySticky cfg.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{sarama.BalanceStrategySticky}
default: default:
return fmt.Errorf("invalid balance strategy %q", k.BalanceStrategy) return fmt.Errorf("invalid balance strategy %q", k.BalanceStrategy)
} }