KafkaConsumer from 0.8.3 trunk hangs indefinitely on poll

["Krogh-Moe, Espen" <es...@finn.no>]

Hi,

I'm trying to test the new java KafkConsumer in 0.8.3 and have pulled the code from trunk, but for me the following just ends up hanging indefinitely.


Map<String, Object> props = new HashMap();
props.put("bootstrap.servers", "broker1:2181");

props.put("key.deserializer», "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("group.id", "test-" + new Random().nextInt());


KafkaConsumer consumer = new KafkaConsumer(props);

consumer.subscribe(topic);

ConsumerRecords<String, V> poller = consumer.poll(100);

poller.forEach(c -> queue.add(c.value()));

I'm able to read from the the topic using kafka-consumer.sh

Is this a know problem, work in progress or my bad in some way or another?

It seems to be hanging while it is trying to update metadata which is for some reason never updated.

Regards
Espen

Re: KafkaConsumer from 0.8.3 trunk hangs indefinitely on poll

[Gwen Shapira <gw...@confluent.io>]

As you can see in the javadoc for KafkaConsumer, you need to call poll() in
a loop. Something like:

while (true) {
*         ConsumerRecords<String, String> records = consumer.poll(100);
*         records.forEach(c -> queue.add(c.value()));
*
*     }


On Tue, Aug 18, 2015 at 2:46 AM, Krogh-Moe, Espen <es...@finn.no>
wrote:

> Hi,
>
> I'm trying to test the new java KafkConsumer in 0.8.3 and have pulled the
> code from trunk, but for me the following just ends up hanging indefinitely.
>
>
> Map<String, Object> props = new HashMap();
> props.put("bootstrap.servers", "broker1:2181");
>
> props.put("key.deserializer»,
> "org.apache.kafka.common.serialization.StringDeserializer");
> props.put("value.deserializer",
> "org.apache.kafka.common.serialization.StringDeserializer");
> props.put("group.id", "test-" + new Random().nextInt());
>
>
> KafkaConsumer consumer = new KafkaConsumer(props);
>
> consumer.subscribe(topic);
>
> ConsumerRecords<String, V> poller = consumer.poll(100);
>
> poller.forEach(c -> queue.add(c.value()));
>
> I'm able to read from the the topic using kafka-consumer.sh
>
> Is this a know problem, work in progress or my bad in some way or another?
>
> It seems to be hanging while it is trying to update metadata which is for
> some reason never updated.
>
> Regards
> Espen
>
>