# Unable to create Kafka consumer from given configuration

**URL:** <https://discuss.elastic.co/t/unable-to-create-kafka-consumer-from-given-configuration/68609>\
**Category:** Logstash\
**Created:** [December 9, 2016, 11:33pm UTC](https://discuss.elastic.co/t/unable-to-create-kafka-consumer-from-given-configuration/68609 "2016-12-09T23:33:30Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![Haomeng](https://avatars.discourse-cdn.com/v4/letter/h/c57346/32.png) [@Haomeng](https://discuss.elastic.co/u/Haomeng)\
**Post date:** [December 9, 2016, 11:33pm UTC](https://discuss.elastic.co/t/unable-to-create-kafka-consumer-from-given-configuration/68609/1 "2016-12-09T23:33:30Z")

</div>

I run with logstash5.0 kafka plugin, got below message, and not sure why is it a ArgumentError, and which argument is wrong, any ideas??

[2016-12-09T16:32:43,420][DEBUG][org.apache.kafka.clients.consumer.KafkaConsumer] Starting the Kafka consumer  
[2016-12-09T16:32:43,420][DEBUG][org.apache.kafka.clients.consumer.KafkaConsumer] The Kafka consumer has closed.  
[2016-12-09T16:32:43,420][ERROR][logstash.inputs.kafka] Unable to create Kafka consumer from given configuration {:kafka\_error\_message=\>org.apache.kafka.common.KafkaException: F to construct kafka consumer}  
[2016-12-09T16:32:43,422][ERROR][logstash.pipeline] A plugin had an unrecoverable error. Will restart this plugin.  
Plugin: \<LogStash::Inputs::Kafka bootstrap\_servers=\>"10.176.95.8", topics=\>["topic1-elasticsearch\_1"], group\_id=\>"logstash\_1", client\_id=\>"logstash\_1", codec=\>LogStash::Codecs::Pid="plain\_176fb102-40e7-4e46-943d-15125f10cb5c", enable\_metric=\>true, charset=\>"UTF-8"\>, consumer\_threads=\>3, decorate\_events=\>true, type=\>"nginx-access", id=\>"94d607dfe60d06ecaedcf4b4ac7413882a4-1", enable\_metric=\>true, auto\_commit\_interval\_ms=\>"5000", enable\_auto\_commit=\>"true", key\_deserializer\_class=\>"org.apache.kafka.common.serialization.StringDeserializvalue\_deserializer\_class=\>"org.apache.kafka.common.serialization.StringDeserializer", ssl=\>false\>  
Error: uncaught throw Failed to construct kafka consumer in thread 0x2e05a  
Exception: ArgumentError  
Stack: org/jruby/RubyKernel.java:1283:in `throw' /usr/share/logstash/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-5.0.5/lib/logstash/inputs/kafka.rb:268:in`create\_consumer'  
/usr/share/logstash/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-5.0.5/lib/logstash/inputs/kafka.rb:179:in `run'

---

<div class="post-metadata">

**Author:** ![Haomeng](https://avatars.discourse-cdn.com/v4/letter/h/c57346/32.png) [@Haomeng](https://discuss.elastic.co/u/Haomeng)\
**Post date:** [December 9, 2016, 11:35pm UTC](https://discuss.elastic.co/t/unable-to-create-kafka-consumer-from-given-configuration/68609/2 "2016-12-09T23:35:47Z")

</div>

Here is my filter argument configuration:

bootstrap\_servers =\> "10.176.95.8"  
topics =\> ["topic1-elasticsearch\_1"]  
group\_id =\> "logstash\_1"  
client\_id =\> "logstash\_1"

---

<div class="post-metadata">

**Author:** ![EddFigueiredo](https://avatars.discourse-cdn.com/v4/letter/e/a587f6/32.png) [@EddFigueiredo](https://discuss.elastic.co/u/EddFigueiredo)\
**Post date:** [December 19, 2016, 8:14pm UTC](https://discuss.elastic.co/t/unable-to-create-kafka-consumer-from-given-configuration/68609/3 "2016-12-19T20:14:40Z")

</div>

I'm having a similar problem, but with logstash version 2.4.1 and kafka\_input\_plugin 6.2.0

{:timestamp=\>"2016-12-19T18:04:05.470000-0200", :message=\>"Flushing buffer at interval", :instance=\>"#\<LogStash::Outputs::ElasticSearch::Buffer:0x68cd91f4 @operations\_mutex=#Mutex:0x660a729d, @max\_size=500, @operations\_lock=#Java::JavaUtilConcurrentLocks::ReentrantLock:0x6235fa84, @submit\_proc=#Proc:0x23f10f85@/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-output-elasticsearch-2.7.1-java/lib/logstash/outputs/elasticsearch/common.rb:57, @logger=#\<Cabin::Channel:0x165a1ffc @metrics=#\<Cabin::Metrics:0x5b763669 @metrics\_lock=#Mutex:0x471887e9, @metrics={}, @channel=#\<Cabin::Channel:0x165a1ffc ...\>\>, @subscriber\_lock=#Mutex:0x6cbe35a8, @level=:debug, @subscribers={239146=\>#\<Cabin::Subscriber:0x4072dc60 @output=#\<Cabin::Outputs::IO:0x43563c48 @io=#\<File:/var/log/logstash/logstash.log\>, @lock=#Mutex:0x5e5852ff\>, @options={}\>, 239148=\>#\<Cabin::Subscriber:0x1fd5079a @output=#\<Cabin::Outputs::IO:0x499249dc @io=#\<IO:fd 1\>, @lock=#Mutex:0x3bd498da\>, @options={:level=\>:fatal}\>}, @data={}\>, @last\_flush=2016-12-19 18:04:04 -0200, @flush\_interval=1, @stopping=#Concurrent::AtomicBoolean:0x10d9d048, @buffer=[], @flush\_thread=#\<Thread:0x4b5ea726 run\>\>", :interval=\>1, :level=\>:debug, :file=\>"logstash/outputs/elasticsearch/buffer.rb", :line=\>"90", :method=\>"interval\_flush"}  
{:timestamp=\>"2016-12-19T18:04:05.493000-0200", :message=\>"Unable to create Kafka consumer from given configuration", :kafka\_error\_message=\>org.apache.kafka.common.KafkaException: Failed to construct kafka consumer, :level=\>:error, :file=\>"logstash/inputs/kafka.rb", :line=\>"315", :method=\>"create\_consumer"}  
{:timestamp=\>"2016-12-19T18:04:05.502000-0200", :message=\>"A plugin had an unrecoverable error. Will restart this plugin.\n Plugin: \<LogStash::Inputs::Kafka codec=\>\<LogStash::Codecs::JSON charset=\>"UTF-8"\>, auto\_commit\_interval\_ms=\>"5000", auto\_offset\_reset=\>"earliest", bootstrap\_servers=\>"server1:9092,server2:9092,server3:9092", client\_id=\>"logstash", connections\_max\_idle\_ms=\>"30000", consumer\_threads=\>2, enable\_auto\_commit=\>"true", fetch\_max\_wait\_ms=\>"500", fetch\_min\_bytes=\>"1000", group\_id=\>"event\_handler", heartbeat\_interval\_ms=\>"10000", max\_partition\_fetch\_bytes=\>"60000000", session\_timeout\_ms=\>"30000", poll\_timeout\_ms=\>600, security\_protocol=\>"SSL", ssl\_truststore\_location=\>"/var/private/ssl/logstash.truststore.jks", ssl\_truststore\_password=\>, ssl\_keystore\_location=\>"/var/private/ssl/logstash.keystore.jks", ssl\_keystore\_password=\>, ssl\_key\_password=\>, topics=\>["zupme-gateway"], key\_deserializer\_class=\>"org.apache.kafka.common.serialization.StringDeserializer", value\_deserializer\_class=\>"org.apache.kafka.common.serialization.StringDeserializer", ssl=\>false, sasl\_mechanism=\>"GSSAPI", decorate\_events=\>false\>\n Error: uncaught throw Failed to construct kafka consumer in thread 0x3a63a\n Exception: ArgumentError\n Stack: org/jruby/RubyKernel.java:1283:in `throw'\n/opt/logstash/vendor/local_gems/27c10342/logstash-input-kafka-6.2.0/lib/logstash/inputs/kafka.rb:316:in`create\_consumer'\n/opt/logstash/vendor/local\_gems/27c10342/logstash-input-kafka-6.2.0/lib/logstash/inputs/kafka.rb:224:in `run'\norg/jruby/RubyFixnum.java:275:in`times'\norg/jruby/RubyEnumerator.java:274:in `each'\norg/jruby/RubyEnumerable.java:757:in`map'\n/opt/logstash/vendor/local\_gems/27c10342/logstash-input-kafka-6.2.0/lib/logstash/inputs/kafka.rb:224:in `run'\n/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-core-2.4.1-java/lib/logstash/pipeline.rb:342:in`inputworker'\n/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-core-2.4.1-java/lib/logstash/pipeline.rb:336:in `start\_input'", :level=\>:error, :file=\>"logstash/pipeline.rb", :line=\>"353", :method=\>"inputworker"}

Apparently it fails to create the consumer thread, but i'm not sure exactly why... frustrating ...

---

<div class="post-metadata">

**Author:** ![Haomeng](https://avatars.discourse-cdn.com/v4/letter/h/c57346/32.png) [@Haomeng](https://discuss.elastic.co/u/Haomeng)\
**Post date:** [December 20, 2016, 2:55am UTC](https://discuss.elastic.co/t/unable-to-create-kafka-consumer-from-given-configuration/68609/4 "2016-12-20T02:55:27Z")

</div>

You can try with below command to probe the configurations, it will show the configuration errors in details:

/usr/share/logstash/bin/logstash --log.level=debug --path.settings=/etc/logstash --config.debug -t

And, a hint, the port number is required for kafka input bootstrap\_servers arguments, so you can add the port if missing to try again, good luck!

bootstrap\_servers =\> "10.176.95.8:9092"

---

<div class="post-metadata">

**Author:** ![EddFigueiredo](https://avatars.discourse-cdn.com/v4/letter/e/a587f6/32.png) [@EddFigueiredo](https://discuss.elastic.co/u/EddFigueiredo)\
**Post date:** [December 20, 2016, 11:36am UTC](https://discuss.elastic.co/t/unable-to-create-kafka-consumer-from-given-configuration/68609/5 "2016-12-20T11:36:31Z")

</div>

I'm my case, I'm using the correct notation host:port, but logstash is still unable to create the consumer group... I had to enable debug mode in logstash to see what was going on, otherwise nothing was shown in the logs.

But still, I don't have a clear reason why it's not working.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [January 17, 2017, 11:36am UTC](https://discuss.elastic.co/t/unable-to-create-kafka-consumer-from-given-configuration/68609/6 "2017-01-17T11:36:43Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.