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 <espen.krogh-...@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
>
>
