Do not add invalid timestamps to kafka messages (#6908)

(cherry picked from commit f6b3026)

plugins/outputs/kafka/kafka.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log"
 	"strings"
+	"time"
 
 	"github.com/Shopify/sarama"
 	"github.com/gofrs/uuid"
@@ -20,6 +21,8 @@ var ValidTopicSuffixMethods = []string{
 	"tags",
 }
 
+var zeroTime = time.Unix(0, 0)
+
 type (
 	Kafka struct {
 		Brokers          []string
@@ -323,9 +326,13 @@ func (k *Kafka) Write(metrics []telegraf.Metric) error {
 		}
 
 		m := &sarama.ProducerMessage{
-			Topic:     k.GetTopicName(metric),
-			Value:     sarama.ByteEncoder(buf),
-			Timestamp: metric.Time(),
+			Topic: k.GetTopicName(metric),
+			Value: sarama.ByteEncoder(buf),
+		}
+
+		// Negative timestamps are not allowed by the Kafka protocol.
+		if !metric.Time().Before(zeroTime) {
+			m.Timestamp = metric.Time()
 		}
 
 		key, err := k.routingKey(metric)