# Kafka Input plugin error

**URL:** <https://discuss.elastic.co/t/kafka-input-plugin-error/31450>\
**Category:** Logstash\
**Created:** [October 1, 2015, 8:11am UTC](https://discuss.elastic.co/t/kafka-input-plugin-error/31450 "2015-10-01T08:11:09Z")\
**Posts on this page:** 9\
**Page:** 1

<div class="post-metadata">

**Author:** ![astro](https://avatars.discourse-cdn.com/v4/letter/a/9dc877/32.png) [@astro](https://discuss.elastic.co/u/astro)\
**Post date:** [October 1, 2015, 8:11am UTC](https://discuss.elastic.co/t/kafka-input-plugin-error/31450/1 "2015-10-01T08:11:09Z")

</div>

Hi All,

I have setup Kafka and trying to use kafka input and output plugin  
For Higher availability I tried to use multiple zookeeper and multiple brokers all setup on different machines .  
zk1:2181 , zk2:2181, zk3:2181 (all zookeeper instance , use zookeeper package from its website)  
kbr1:9092, kbr2:9092 ( kafka broker instance )

Following Error is coming

kafka client threw exception, restarting {:exception=\>java.net.UnknownHostException: logstash2: logstash2: unknown error, :level=\>:warn, :file=\>"logstash/inputs/kafka.rb", :line=\>"150", :method=\>"run"}  
^CSIGINT received. Shutting down the pipeline. {:level=\>:warn, :file=\>"logstash/agent.rb", :line=\>"126", :method=\>"execute"}  
log4j, [2015-10-01T13:30:43.766] DEBUG: org.I0Itec.zkclient.ZkConnection: Creating new ZookKeeper instance to connect to zk1:2181,zk2:2181:2181,zk3:2181.  
log4j, [2015-10-01T13:30:43.766] INFO: org.apache.zookeeper.ZooKeeper: Initiating client connection, connectString=zk1:2181,zk2:2181:2181,zk3:2181 sessionTimeout=30000 watcher=org.I0Itec.zkclient.ZkClient@1e472620  
log4j, [2015-10-01T13:30:43.766] DEBUG: org.I0Itec.zkclient.ZkClient: Closing ZkClient...  
log4j, [2015-10-01T13:30:43.767] INFO: org.I0Itec.zkclient.ZkEventThread: Starting ZkClient event thread.  
log4j, [2015-10-01T13:30:43.768] DEBUG: org.I0Itec.zkclient.ZkClient: Closing ZkClient...done  
log4j, [2015-10-01T13:30:43.768] INFO: kafka.utils.VerifiableProperties: Verifying properties  
log4j, [2015-10-01T13:30:43.769] INFO: kafka.utils.VerifiableProperties: Property auto.commit.enable is overridden to true  
log4j, [2015-10-01T13:30:43.769] INFO: kafka.utils.VerifiableProperties: Property auto.commit.interval.ms is overridden to 1000  
log4j, [2015-10-01T13:30:43.769] INFO: kafka.utils.VerifiableProperties: Property auto.offset.reset is overridden to largest  
log4j, [2015-10-01T13:30:43.769] INFO: kafka.utils.VerifiableProperties: Property consumer.timeout.ms is overridden to -1  
log4j, [2015-10-01T13:30:43.769] INFO: kafka.utils.VerifiableProperties: Property fetch.message.max.bytes is overridden to 1048576  
log4j, [2015-10-01T13:30:43.769] INFO: kafka.utils.VerifiableProperties: Property fetch.min.bytes is overridden to 1  
log4j, [2015-10-01T13:30:43.769] INFO: kafka.utils.VerifiableProperties: Property fetch.wait.max.ms is overridden to 100  
log4j, [2015-10-01T13:30:43.770] INFO: kafka.utils.VerifiableProperties: Property group.id is overridden to t  
log4j, [2015-10-01T13:30:43.770] INFO: kafka.utils.VerifiableProperties: Property queued.max.message.chunks is overridden to 10  
log4j, [2015-10-01T13:30:43.770] INFO: kafka.utils.VerifiableProperties: Property rebalance.backoff.ms is overridden to 2000  
log4j, [2015-10-01T13:30:43.770] INFO: kafka.utils.VerifiableProperties: Property rebalance.max.retries is overridden to 4  
log4j, [2015-10-01T13:30:43.770] INFO: kafka.utils.VerifiableProperties: Property refresh.leader.backoff.ms is overridden to 200  
log4j, [2015-10-01T13:30:43.770] INFO: kafka.utils.VerifiableProperties: Property socket.receive.buffer.bytes is overridden to 65536  
log4j, [2015-10-01T13:30:43.770] INFO: kafka.utils.VerifiableProperties: Property socket.timeout.ms is overridden to 30000  
log4j, [2015-10-01T13:30:43.771] INFO: kafka.utils.VerifiableProperties: Property zookeeper.connect is overridden to [logstash1.shopclues.com:2181](http://logstash1.shopclues.com:2181),logstash2.shopclues.com:2181:2181,[logstash3.shopclues.com:2181](http://logstash3.shopclues.com:2181)  
log4j, [2015-10-01T13:30:43.771] INFO: kafka.utils.VerifiableProperties: Property zookeeper.connection.timeout.ms is overridden to 6000  
log4j, [2015-10-01T13:30:43.771] INFO: kafka.utils.VerifiableProperties: Property zookeeper.session.timeout.ms is overridden to 6000  
log4j, [2015-10-01T13:30:43.771] INFO: kafka.utils.VerifiableProperties: Property zookeeper.sync.time.ms is overridden to 2000  
Sending shutdown signal to input thread {:thread=\>#\<Thread:0x7b558db9 sleep\>, :level=\>:info, :file=\>"logstash/pipeline.rb", :line=\>"260", :method=\>"shutdown"}  
Plugin is finished {:plugin=\>\<LogStash::Inputs::Kafka zk\_connect=\>"zk1:2181,zk2:2181:2181,zk3:2181", group\_id=\>"t", topic\_id=\>"test", reset\_beginning=\>true, consumer\_threads=\>1, consumer\_restart\_on\_error=\>true, consumer\_restart\_sleep\_ms=\>100, decorate\_events=\>true, debug=\>false, codec=\>\<LogStash::Codecs::JSON charset=\>"UTF-8"\>, auto\_offset\_reset=\>"largest", queue\_size=\>20, rebalance\_max\_retries=\>4, rebalance\_backoff\_ms=\>2000, consumer\_timeout\_ms=\>-1, fetch\_message\_max\_bytes=\>1048576, decoder\_class=\>"kafka.serializer.DefaultDecoder", key\_decoder\_class=\>"kafka.serializer.DefaultDecoder"\>, :level=\>:info, :file=\>"logstash/plugin.rb", :line=\>"61", :method=\>"finished"}  
Plugin is finished {:plugin=\>\<LogStash::Outputs::Stdout codec=\>\<LogStash::Codecs::RubyDebug metadata=\>false\>, workers=\>1\>, :level=\>:info, :file=\>"logstash/plugin.rb", :line=\>"61", :method=\>"finished"}  
Pipeline shutdown complete. {:level=\>:info, :file=\>"logstash/pipeline.rb", :line=\>"101", :method=\>"run"}

---

<div class="post-metadata">

**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\
**Post date:** [October 1, 2015, 12:55pm UTC](https://discuss.elastic.co/t/kafka-input-plugin-error/31450/2 "2015-10-01T12:55:36Z")

</div>

Make sure the host.name property on the brokers is something resolvable by  
the consumers.

---

<div class="post-metadata">

**Author:** ![astro](https://avatars.discourse-cdn.com/v4/letter/a/9dc877/32.png) [@astro](https://discuss.elastic.co/u/astro)\
**Post date:** [October 1, 2015, 1:08pm UTC](https://discuss.elastic.co/t/kafka-input-plugin-error/31450/3 "2015-10-01T13:08:08Z")

</div>

@Joe_Lawson

I am still receiving the same error , host.name is set to kb1 and kb2 and both server entries are there is /etc/hosts file on the consumer

---

<div class="post-metadata">

**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\
**Post date:** [October 1, 2015, 1:30pm UTC](https://discuss.elastic.co/t/kafka-input-plugin-error/31450/4 "2015-10-01T13:30:13Z")

</div>

Can you post your logstash config?

---

<div class="post-metadata">

**Author:** ![astro](https://avatars.discourse-cdn.com/v4/letter/a/9dc877/32.png) [@astro](https://discuss.elastic.co/u/astro)\
**Post date:** [October 1, 2015, 1:57pm UTC](https://discuss.elastic.co/t/kafka-input-plugin-error/31450/5 "2015-10-01T13:57:57Z")

</div>

@Joe_Lawson

Here's the config

input {  
kafka {  
zk\_connect =\> "zk1:2181,zk2:2181:2181,zk3:2181"  
group\_id =\> "t"  
topic\_id =\> "test"  
reset\_beginning =\> true  
consumer\_threads =\> 1  
consumer\_restart\_on\_error =\> false  
consumer\_restart\_sleep\_ms =\> 100  
decorate\_events =\> true  
}  
}

output { stdout { codec =\> rubydebug} }

---

<div class="post-metadata">

**Author:** ![astro](https://avatars.discourse-cdn.com/v4/letter/a/9dc877/32.png) [@astro](https://discuss.elastic.co/u/astro)\
**Post date:** [October 3, 2015, 7:49am UTC](https://discuss.elastic.co/t/kafka-input-plugin-error/31450/6 "2015-10-03T07:49:49Z")

</div>

@Joe_Lawson is there something that can be done in this case

---

<div class="post-metadata">

**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\
**Post date:** [October 6, 2015, 11:55pm UTC](https://discuss.elastic.co/t/kafka-input-plugin-error/31450/7 "2015-10-06T23:55:30Z")

</div>

Seems like logstash2 is an unknown host to the consumer. Is that the  
hostname of something?  
astro [http://discuss.elastic.co/users/astro](http://discuss.elastic.co/users/astro) Arvind Sharma  
[http://discuss.elastic.co/users/astro](http://discuss.elastic.co/users/astro)  
October 3

@Joe_Lawson [http://discuss.elastic.co/users/joe\_lawson](http://discuss.elastic.co/users/joe_lawson) is there something  
that can be done in this case

---

<div class="post-metadata">

**Author:** ![astro](https://avatars.discourse-cdn.com/v4/letter/a/9dc877/32.png) [@astro](https://discuss.elastic.co/u/astro)\
**Post date:** [October 9, 2015, 1:30pm UTC](https://discuss.elastic.co/t/kafka-input-plugin-error/31450/8 "2015-10-09T13:30:36Z")

</div>

@Joe_Lawson

I figured it out , the logstash consumer for kafka was getting hostname as logtash2 from /etc/sysconfig/network but in /etc/hosts file it was set as ls2 for 127.0.0.1 , also I added one more thing i.e the kafka logstash consumers should resolve both zookeeper hosts as well as kafka broker hosts so I added entry for both in /etc/hosts file ( earlier I made entry for zk hosts only )

btw thanks

---

<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:** [July 6, 2017, 5:27am UTC](https://discuss.elastic.co/t/kafka-input-plugin-error/31450/9 "2017-07-06T05:27:01Z")

</div>