# Logstash subscribe Kafka message gets ClosedChannelException

**URL:** <https://discuss.elastic.co/t/logstash-subscribe-kafka-message-gets-closedchannelexception/31445>\
**Category:** Logstash\
**Created:** [October 1, 2015, 7:39am UTC](https://discuss.elastic.co/t/logstash-subscribe-kafka-message-gets-closedchannelexception/31445 "2015-10-01T07:39:45Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![Jason\_Zheng](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jason_zheng/32/4041_2.png) [@Jason\_Zheng](https://discuss.elastic.co/u/Jason_Zheng)\
**Post date:** [October 1, 2015, 7:39am UTC](https://discuss.elastic.co/t/logstash-subscribe-kafka-message-gets-closedchannelexception/31445/1 "2015-10-01T07:39:45Z")

</div>

Hi All,

I would like to get message from kafka but get ClosedChannelException (loops)

```
log4j, [2015-10-01T15:44:28.569] WARN: kafka.consumer.ConsumerFetcherManager$LeaderFinderThread: [logstash_logstashsubkafka-1443685462254-3d51c80d-leader-finder-thread], Failed to find leader for Set([neptunus,0])
kafka.common.KafkaException: fetching topic metadata for topics [Set(neptunus)] from broker [ArrayBuffer(id:1,host:kafkahost,port:9093)] failed
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:72)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:93)
        at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
        at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60)
Caused by: java.nio.channels.ClosedChannelException
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
        at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:113)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)
        ... 3 more

```

Kafka

```
    ./kafka-console-consumer.sh --zookeeper zookeeper4kafka:2181 --topic neptunus --from-beginning
    {"message":"helloWorld!!","@version":"1","@timestamp":"2015-10-01T04:14:36.579Z","host":"logstashpubkafka","path":"/var/log/kafka/test","tst"}
    {"message":"Test World","@version":"1","@timestamp":"2015-10-01T04:20:34.966Z","host":"logstashpubkafka","path":"/var/log/kafka/test","typ"}
    {"message":"helloWorld!!","@version":"1","@timestamp":"2015-10-01T04:20:34.968Z","host":"logstashpubkafka","path":"/var/log/kafka/test","tst"}
    {"message":"today is my day","@version":"1","@timestamp":"2015-10-01T05:42:04.827Z","host":"logstashpubkafka","path":"/var/log/kafka/test""test"}

```

Logstash (subscribe)

```
input {
        kafka {
                topic_id => "neptunus"
                zk_connect => "zookeeper4kafka:2181"
                type => "test"
        }
}

output {
        stdout { codec => rubydebug }
}

```

Any helps

Jason

---

<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:45pm UTC](https://discuss.elastic.co/t/logstash-subscribe-kafka-message-gets-closedchannelexception/31445/2 "2015-10-01T12:45:31Z")

</div>

Make sure the Kafka brokers have their host.name property set to something  
resolvable by the consumer.

---

<div class="post-metadata">

**Author:** ![Jason\_Zheng](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jason_zheng/32/4041_2.png) [@Jason\_Zheng](https://discuss.elastic.co/u/Jason_Zheng)\
**Post date:** [October 1, 2015, 2:52pm UTC](https://discuss.elastic.co/t/logstash-subscribe-kafka-message-gets-closedchannelexception/31445/3 "2015-10-01T14:52:22Z")

</div>

Hi Joe,

Thanks for replying, I solved the problem by your recommended

Jason

---

<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/logstash-subscribe-kafka-message-gets-closedchannelexception/31445/4 "2017-07-06T05:27:31Z")

</div>