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

fix(outputs.kafka): Simplify send-error handling (#14154)

This commit is contained in:
Sven Rebhan 2023-10-23 15:51:57 +02:00 committed by GitHub
parent 184038d5e6
commit fd773b3e28
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 14 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
plugins/outputs/kafka/kafka.go
View File

@ -223,21 +223,21 @@ func (k *Kafka) Write(metrics []telegraf.Metric) error {
if err != nil {
// We could have many errors, return only the first encountered.
var errs sarama.ProducerErrors
if errors.As(err, &errs) {
for _, prodErr := range errs {
if errors.Is(prodErr.Err, sarama.ErrMessageSizeTooLarge) {
k.Log.Error("Message too large, consider increasing `max_message_bytes`; dropping batch")
return nil
}
if errors.Is(prodErr.Err, sarama.ErrInvalidTimestamp) {
k.Log.Error(
"The timestamp of the message is out of acceptable range, consider increasing broker `message.timestamp.difference.max.ms`; " +
"dropping batch",
)
return nil
}
return prodErr //nolint:staticcheck // Return first error encountered
if errors.As(err, &errs) && len(errs) > 0 {
// Just return the first error encountered
firstErr := errs[0]
if errors.Is(firstErr.Err, sarama.ErrMessageSizeTooLarge) {
k.Log.Error("Message too large, consider increasing `max_message_bytes`; dropping batch")
return nil
}
if errors.Is(firstErr.Err, sarama.ErrInvalidTimestamp) {
k.Log.Error(
"The timestamp of the message is out of acceptable range, consider increasing broker `message.timestamp.difference.max.ms`; " +
"dropping batch",
)
return nil
}
return firstErr
}
return err
}