kafka input problematic consumer threads handling

[colinsurprenant]

the input plugin simply spans a number of consumer threads and join them in the run method:

logstash-integration-kafka/lib/logstash/inputs/kafka.rb

Lines 227 to 228 in d6530c0

	@runner_threads = @runner_consumers.map { |consumer| thread_runner(logstash_queue, consumer) }
	@runner_threads.each { |t| t.join }

This leads to 2 potentially serious problems:

• These consumer threads are unmanaged and could silently die depending on the exception and if the dying thread is the one join'ed or not.
• An exception in a consumer thread can crash logstash see input workers exception handling elastic/logstash#11603

[RogierSchuring]

Had this exact issue last week.
I expected only the pipeline with the Kafka cluster that was faulty to restart or fail, but was unpleasantly surprised that the whole Logstash agent restarts.

In our case it was expired certificates on some Kafka nodes of one Kafka cluster that caused handshake failures.

Would love a fix for this to prevent this in the future.