Re: Consumer blocks forever on unreachable ports

[Raghu Angadi <an...@gmail.com>]

quick update: This can be unblocked with consumer.wakeup(). So my current
work around is to run this in a separate thread and cancel it after a
timeout.

On Fri, Jun 30, 2017 at 11:14 AM, Raghu Angadi <an...@gmail.com> wrote:

> 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)
>
>