Kafka High Level Consumer Fetch All Messages From Topic Using Java API (Equivalent to --from-beginning)

[jp...@yahoo.com]

Good Morning,

I am testing the Kafka High Level Consumer using the ConsumerGroupExample code from the Kafka site. I would like to retrieve all the existing messages on the topic called "test" that I have in the Kafka server config. Looking at other blogs, auto.offset.reset should be set to "smallest" to be able to get all messages:
private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId)    { Properties props = new Properties(); props.put("zookeeper.connect", a_zookeeper); props.put("group.id", a_groupId); props.put("auto.offset.reset", "smallest"); props.put("zookeeper.session.timeout.ms", "10000");      return new ConsumerConfig(props);
}
The question I really have is this: what is the equivalent Java api call for the High Level Consumer that is the equivalent of:
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
Thx for your help!!

Re: Kafka High Level Consumer Fetch All Messages From Topic Using Java API (Equivalent to --from-beginning)

[pushkar priyadarshi <pr...@gmail.com>]

I don't think there is any direct high level API equivalent to this.every
time you read messages using high level api your offset gets synced in zoo
keeper .auto offset is for cases where last read offset for example have
been purged n rather than getting exception you want to just fall back to
either most current or oldest message offset.
But other's more experienced opinion on this will be great.
Regards,
Pushkar
On Feb 14, 2014 4:40 PM, <jp...@yahoo.com> wrote:

> Good Morning,
>
> I am testing the Kafka High Level Consumer using the ConsumerGroupExample
> code from the Kafka site. I would like to retrieve all the existing
> messages on the topic called "test" that I have in the Kafka server config.
> Looking at other blogs, auto.offset.reset should be set to "smallest" to be
> able to get all messages:
> private static ConsumerConfig createConsumerConfig(String a_zookeeper,
> String a_groupId)    { Properties props = new Properties();
> props.put("zookeeper.connect", a_zookeeper); props.put("group.id",
> a_groupId); props.put("auto.offset.reset", "smallest"); props.put("
> zookeeper.session.timeout.ms", "10000");      return new
> ConsumerConfig(props);
> }
> The question I really have is this: what is the equivalent Java api call
> for the High Level Consumer that is the equivalent of:
> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test
> --from-beginning
> Thx for your help!!