Consumer blocks forever on unreachable ports

[Raghu Angadi]

Consumer blocks forever during initialization if the brokers are not
reachable. 'request.timeout.ms' does not seem to be effective. How are
users expected to detect this and handle?

e.g:
$ ./kafka-console-consumer.sh --bootstrap-server google.com:9092 --topic
queries

This blocks forever since google.com:9092 is not reachable. Consumer stack
trace :

[...]
        - locked <0x00000000fdcba750> (a sun.nio.ch.Util$3)
        - locked <0x00000000fdcba740> (a
java.util.Collections$UnmodifiableSet)
        - locked <0x00000000fdcba618> (a sun.nio.ch.EPollSelectorImpl)
        at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
        at
org.apache.kafka.common.network.Selector.select(Selector.java:470)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:286)
        at
org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)
        at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:232)
        - locked <0x00000000fdceacd0> (a
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient)
        at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:209)
        at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:148)
        at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:136)
        at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:197)
        - locked <0x00000000fdcea350> (a
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
        at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:248)
        at
org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1013)
        at
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:979)
        at kafka.consumer.NewShinyConsumer.<init>(BaseConsumer.scala:67)
        at kafka.tools.ConsoleConsumer$.run(ConsoleConsumer.scala:69)
        at kafka.tools.ConsoleConsumer$.main(ConsoleConsumer.scala:50)
        at kafka.tools.ConsoleConsumer.main(ConsoleConsumer.scala)