From 193aba86735dfa8859be8f9db321b5b0f0fbe165 Mon Sep 17 00:00:00 2001 From: Daniel Nelson Date: Tue, 15 Jan 2019 11:48:36 -0800 Subject: [PATCH] Ack delivery if it is unparseable in amqp_consumer input (#5286) --- plugins/inputs/amqp_consumer/amqp_consumer.go | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/plugins/inputs/amqp_consumer/amqp_consumer.go b/plugins/inputs/amqp_consumer/amqp_consumer.go index 568ee6f38..d80a3683b 100644 --- a/plugins/inputs/amqp_consumer/amqp_consumer.go +++ b/plugins/inputs/amqp_consumer/amqp_consumer.go @@ -430,6 +430,14 @@ func (a *AMQPConsumer) process(ctx context.Context, msgs <-chan amqp.Delivery, a func (a *AMQPConsumer) onMessage(acc telegraf.TrackingAccumulator, d amqp.Delivery) error { metrics, err := a.parser.Parse(d.Body) if err != nil { + // Discard the message from the queue; will never be able to process + // this message. + rejErr := d.Ack(false) + if rejErr != nil { + log.Printf("E! [inputs.amqp_consumer] Unable to reject message: %d: %v", + d.DeliveryTag, rejErr) + a.conn.Close() + } return err }