Mailing-List: contact users-help@kafka.apache.org; run by ezmlm
Precedence: bulk
Reply-To: users@kafka.apache.org
Received-SPF: pass (athena.apache.org: domain of achanta.vamsi@flipkart.com
 designates 209.85.160.175 as permitted sender)
MIME-Version: 1.0
In-Reply-To: 
 <CAHXdro=fBiWfRPq+3KzpoFmqd9rjsMCizRrekVW_RUX3BD1K=w@mail.gmail.com>
References: 
 <CAHXdro=fBiWfRPq+3KzpoFmqd9rjsMCizRrekVW_RUX3BD1K=w@mail.gmail.com>
Date: Tue, 17 Jun 2014 14:54:29 +0530
Message-ID: 
 <CAHXdro=hXxDPkZo8caRNbnREvZQnGQNod1uQy=J9gVna60eWqA@mail.gmail.com>
Subject: Re: Offset of last un-consumed message
From: Achanta Vamsi Subhash <achanta.vamsi@flipkart.com>
To: users@kafka.apache.org
Content-Type: multipart/alternative; boundary=20cf303a2d61ef4a1104fc04b5b3

--20cf303a2d61ef4a1104fc04b5b3
Content-Type: text/plain; charset=UTF-8

Sorry. I want the first un-consumed message offset.


On Tue, Jun 17, 2014 at 2:53 PM, Achanta Vamsi Subhash <
achanta.vamsi@flipkart.com> wrote:

> Hi,
>
> I have a consumer group with multiple threads (high-level consumers) which
> read from a topic.
>
> I am also using a SimpleConsumer to read messages given a start offset. I
> am getting the offset as the last produced message using the below code.
> How to get the last un-consumed message?
>
>     public long getLastOffset(SimpleConsumer consumer, String topic, int
> partition,
>                                      long whichTime, String clientName) {
>         TopicAndPartition topicAndPartition = new TopicAndPartition(topic,
> partition);
>         Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo =
> new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
>         requestInfo.put(topicAndPartition, new
> PartitionOffsetRequestInfo(whichTime, maxReads));
>         kafka.javaapi.OffsetRequest request = new
> kafka.javaapi.OffsetRequest(
>                 requestInfo, kafka.api.OffsetRequest.CurrentVersion(),
> clientName);
>         OffsetResponse response = consumer.getOffsetsBefore(request);
>
>         if (response.hasError()) {
>             LOGGER.error("Error fetching data Offset Data the Broker.
> Reason: " + response.errorCode(topic, partition) );
>             return 0;
>         }
>         long[] offsets = response.offsets(topic, partition);
>         return offsets[0];
>     }
>
>
> --
> Regards
> Vamsi Subhash
>


-- 
Regards
Vamsi Subhash

--20cf303a2d61ef4a1104fc04b5b3--