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