Make kafka client unique across multiple pipelines

Author: w32-blaster

lib/logstash/inputs/kafka.rb

@@ -131,6 +131,8 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base
   # The class name of the partition assignment strategy that the client will use to distribute
   # partition ownership amongst consumer instances
   config :partition_assignment_strategy, :validate => :string
+  # ID of the pipeline whose events you want to read from.
+  config :pipeline_id, :validate => :string, :default => "main"
   # The size of the TCP receive buffer (SO_RCVBUF) to use when reading data.
   config :receive_buffer_bytes, :validate => :string
   # The amount of time to wait before attempting to reconnect to a given host.
@@ -221,7 +223,7 @@ def register
 
   public
   def run(logstash_queue)
-    @runner_consumers = consumer_threads.times.map { |i| create_consumer("#{client_id}-#{i}") }
+    @runner_consumers = consumer_threads.times.map { |i| create_consumer("#{client_id}-#{i}-#{pipeline_id}") }
     @runner_threads = @runner_consumers.map { |consumer| thread_runner(logstash_queue, consumer) }
     @runner_threads.each { |t| t.join }
   end # def run

logstash-input-kafka.iml

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module external.linked.project.id="logstash-input-kafka" external.linked.project.path="$MODULE_DIR$" external.root.project.path="$MODULE_DIR$" external.system.id="GRADLE" type="JAVA_MODULE" version="4">
+  <component name="NewModuleRootManager" inherit-compiler-output="true">
+    <exclude-output />
+    <content url="file://$MODULE_DIR$" />
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>

spec/integration/inputs/kafka_spec.rb

@@ -13,7 +13,7 @@
   let(:group_id_4) {rand(36**8).to_s(36)}
   let(:group_id_5) {rand(36**8).to_s(36)}
   let(:plain_config) { { 'topics' => ['logstash_topic_plain'], 'codec' => 'plain', 'group_id' => group_id_1, 'auto_offset_reset' => 'earliest'} }
-  let(:multi_consumer_config) { plain_config.merge({"group_id" => group_id_4, "client_id" => "spec", "consumer_threads" => 3}) }
+  let(:multi_consumer_config) { plain_config.merge({"group_id" => group_id_4, "client_id" => "spec", "consumer_threads" => 3, "pipeline_id" => "spec_pipeline"}) }
   let(:snappy_config) { { 'topics' => ['logstash_topic_snappy'], 'codec' => 'plain', 'group_id' => group_id_1, 'auto_offset_reset' => 'earliest'} }
   let(:lz4_config) { { 'topics' => ['logstash_topic_lz4'], 'codec' => 'plain', 'group_id' => group_id_1, 'auto_offset_reset' => 'earliest'} }
   let(:pattern_config) { { 'topics_pattern' => 'logstash_topic_.*', 'group_id' => group_id_2, 'codec' => 'plain', 'auto_offset_reset' => 'earliest'} }  
@@ -82,7 +82,7 @@ def thread_it(kafka_input, queue)
         wait(timeout_seconds).for {queue.length}.to eq(num_events)
         expect(queue.length).to eq(num_events)
         kafka_input.kafka_consumers.each_with_index do |consumer, i|
-          expect(consumer.metrics.keys.first.tags["client-id"]).to eq("spec-#{i}")
+          expect(consumer.metrics.keys.first.tags["client-id"]).to eq("spec-#{i}-spec_pipeline")
         end
       ensure
         t.kill

spec/unit/inputs/kafka_spec.rb

@@ -30,7 +30,8 @@ def wakeup
 end
 
 describe LogStash::Inputs::Kafka do
-  let(:config) { { 'topics' => ['logstash'], 'consumer_threads' => 4 } }
+  let(:pipeline_id) { SecureRandom.hex(8)}
+  let(:config) { { 'topics' => ['logstash'], 'consumer_threads' => 4, 'pipeline_id' => pipeline_id } }
   subject { LogStash::Inputs::Kafka.new(config) }
 
   it "should register" do