Multiple Consumer Threads Share a Single Instance of the @codec

[original-brownbear]

The @codec is not threadsafe and we're one instance of it across threads here:

private
  def thread_runner(logstash_queue, consumer)
    Thread.new do
      begin
        unless @topics_pattern.nil?
          nooplistener = org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener.new
          pattern = java.util.regex.Pattern.compile(@topics_pattern)
          consumer.subscribe(pattern, nooplistener)
        else
          consumer.subscribe(topics);
        end
        while !stop?
          records = consumer.poll(poll_timeout_ms)
          for record in records do
            @codec.decode(record.value.to_s) do |event|

we must clone the @codec once per thread like we do in tcp input for example to avoid concurrency issues.

[original-brownbear]

fixed in #211