title: Kafka throws java.nio.channels.ClosedChannelException
body: <p>When i try to consume messages from the kafka server which is hosted in ec2 with kafka console tool (V 0.9.0.1 , i think this uses old consumer APIs)
  I get following exception.
  How can i overcome this?</p>
  
  <p><strong>#./kafka-console-consumer.sh --zookeeper zookeeper1.xx.com:2181 --topic MY_TOPIC --from-beginning</strong></p>
  
  <pre><code>[2016-04-06 14:34:58,219] WARN Fetching topic metadata with correlation id 0 for topics [Set(MY_TOPIC)] from broker [BrokerEndPoint(1014,kafka3.xx.com,9092)] failed (kafka.client.ClientUtils$)
  java.nio.channels.ClosedChannelException
      at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
      at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
      at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74)
      at kafka.producer.SyncProducer.send(SyncProducer.scala:119)
      at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
      at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
      at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
      at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
  [2016-04-06 14:34:58,222] WARN Fetching topic metadata with correlation id 0 for topics [Set(MY_TOPIC)] from broker [BrokerEndPoint(1013,kafka22.xx.com,9092)] failed (kafka.client.ClientUtils$)
  java.nio.channels.ClosedChannelException
      at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
      at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
      at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74)
      at kafka.producer.SyncProducer.send(SyncProducer.scala:119)
      at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
      at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
      at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
      at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
  [2016-
  </code></pre>