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

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

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

--20cf304275c8d065fc04fc04b3b5--