Send all messages before waiting for results in kafka output (#4491)

This commit is contained in:
Daniel Nelson 2018-07-31 15:08:04 -07:00 committed by Greg
parent 93ed28e745
commit 943dcc0c49
1 changed files with 16 additions and 14 deletions

View File

@ -246,32 +246,34 @@ func (k *Kafka) Description() string {
} }
func (k *Kafka) Write(metrics []telegraf.Metric) error { func (k *Kafka) Write(metrics []telegraf.Metric) error {
if len(metrics) == 0 { msgs := make([]*sarama.ProducerMessage, 0, len(metrics))
return nil
}
for _, metric := range metrics { for _, metric := range metrics {
buf, err := k.serializer.Serialize(metric) buf, err := k.serializer.Serialize(metric)
if err != nil { if err != nil {
return err return err
} }
topicName := k.GetTopicName(metric)
m := &sarama.ProducerMessage{ m := &sarama.ProducerMessage{
Topic: topicName, Topic: k.GetTopicName(metric),
Value: sarama.ByteEncoder(buf), Value: sarama.ByteEncoder(buf),
} }
if h, ok := metric.Tags()[k.RoutingTag]; ok { if h, ok := metric.GetTag(k.RoutingTag); ok {
m.Key = sarama.StringEncoder(h) m.Key = sarama.StringEncoder(h)
} }
msgs = append(msgs, m)
}
_, _, err = k.producer.SendMessage(m) err := k.producer.SendMessages(msgs)
if err != nil { if err != nil {
return fmt.Errorf("FAILED to send kafka message: %s\n", err) // We could have many errors, return only the first encountered.
if errs, ok := err.(sarama.ProducerErrors); ok {
for _, prodErr := range errs {
return prodErr
} }
} }
return err
}
return nil return nil
} }