From f88cffa782440e45da9b31bf8951f99278444aae Mon Sep 17 00:00:00 2001
From: Tobias Jungel <tobias.jungel@gmail.com>
Date: Thu, 7 Sep 2023 18:23:05 +0200
Subject: [PATCH] fix(inputs.kafka_consumer): use mutex for parser

Using parsers like json_v2 will result in undesired parser results. This
adds a mutex to calls to the parser to prevent side effects from calling
the parser in parallel.
---
 plugins/inputs/kafka_consumer/kafka_consumer.go | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/plugins/inputs/kafka_consumer/kafka_consumer.go b/plugins/inputs/kafka_consumer/kafka_consumer.go
index 2d1d5901396df..1a24363d030a6 100644
--- a/plugins/inputs/kafka_consumer/kafka_consumer.go
+++ b/plugins/inputs/kafka_consumer/kafka_consumer.go
@@ -456,7 +456,9 @@ func (h *ConsumerGroupHandler) Handle(session sarama.ConsumerGroupSession, msg *
 			len(msg.Value), h.MaxMessageLen)
 	}
 
+	h.mu.Lock()
 	metrics, err := h.parser.Parse(msg.Value)
+	h.mu.Unlock()
 	if err != nil {
 		h.release()
 		return err