#210 clone codec once per input thread since it is not threadsafe

Fixes #211

Author: original-brownbear

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 6.3.4
+  - Fix an issue that led to random failures in decoding messages when using more than one input thread
+
 ## 6.3.3
   - Upgrade Kafka client to version 0.11.0.0
 

lib/logstash/inputs/kafka.rb

@@ -247,10 +247,11 @@ def thread_runner(logstash_queue, consumer)
         else
           consumer.subscribe(topics);
         end
+        codec_instance = @codec.clone
         while !stop?
-          records = consumer.poll(poll_timeout_ms);
+          records = consumer.poll(poll_timeout_ms)
           for record in records do
-            @codec.decode(record.value.to_s) do |event|
+            codec_instance.decode(record.value.to_s) do |event|
               decorate(event)
               if @decorate_events
                 event.set("[kafka][topic]", record.topic)

logstash-input-kafka.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-input-kafka'
-  s.version         = '6.3.3'
+  s.version         = '6.3.4'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = 'This input will read events from a Kafka topic. It uses the high level consumer API provided by Kafka to read messages from the broker'
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"