Java API offset operations clarification

[Marina <pp...@yahoo.com.INVALID>]

Hi,I'mtrying to use low-level Consumer  JavaAPI to manage offsets manually, with the latest kafka_2.10-0.8.2.1To verify that theoffsets I commit/read form Kafka are correct, I use thekafka.tools.ConsumerOffsetChecker tool.Here is an exampleof the output for a topic/consumer group ./bin/kafka-run-class.shkafka.tools.ConsumerOffsetChecker --group elastic_search_group --zookeeperlocalhost:2181 --topic my_log_topicGroup          Topic                         Pid Offset         logSize         Lag            Ownerelastic_search_groupmy_log_topic              0   5              29             24              none Hereis my interpretation of the result:Offset= 5 --> this is the current offset of my 'elastic_search_group' consumerlogSize= 29 --> this is the Latest offset - the offset of the next message thatwill come to this topic/partitionLag= 24  --> 29-5 - how many messages arenot yet processed by my 'elastic_search_group' consumer Q1:is my interpretation correct? And what is the Pid in this case? Now,I want to get the same information from my Java consumer. Here, I found that Ihad to use two different APIs:kafka.javaapi.OffsetRequestto get Earliest and Latest offsets, but kafka.javaapi.OffsetFetchRequest to getthe current offset. To get the Earliest (orLatest) offset I do:
TopicAndPartition topicAndPartition = newTopicAndPartition(myTopic, myPartition);Map<TopicAndPartition,PartitionOffsetRequestInfo> requestInfo =new HashMap<TopicAndPartition,PartitionOffsetRequestInfo>();requestInfo.put(topicAndPartition,newPartitionOffsetRequestInfo(OffsetRequest.EarliestTime(), 1));// OR for Latest: requestInfo.put(topicAndPartition,newPartitionOffsetRequestInfo(OffsetRequest.LatestTime(), 1));kafka.javaapi.OffsetRequestrequest = newkafka.javaapi.OffsetRequest(requestInfo,kafka.api.OffsetRequest.CurrentVersion(), clientName);OffsetResponse response = simpleConsumer.getOffsetsBefore(request);long[] offsets = response.offsets(topic, partition);long myEarliestOffset = offsets[0];// OR for Latest: long myLatestOffset = offsets[0];
 And to get thecurrent offset I have to use a completely different API:
short versionID = 0;int correlationId = 0;List<TopicAndPartition>topicPartitionList = new ArrayList<TopicAndPartition>(); TopicAndPartition myTopicAndPartition = newTopicAndPartition(myTopic, myPartition);topicPartitionList.add(myTopicAndPartition); OffsetFetchRequest offsetFetchReq = newOffsetFetchRequest(kafkaGroupId, topicPartitionList, versionID, correlationId, kafkaGroupId);OffsetFetchResponse offsetFetchResponse = simpleConsumer.fetchOffsets(offsetFetchReq);long currentOffset = offsetFetchResponse.offsets().get(myTopicAndPartition).offset();
 Q2: is it correct?why are there two different APIs to get a very similar information? Q3: does it matterwhich versionId and correlationId I am using here? I though  versionId should be 0 for pre-0.8.2.1 kafka,and be 1 for 0.8.2.1 and later - but seems like it works with 0 for 0.8.2.1 aswell - see below ? So,for the the example state of the topic above, and the above output of the ConsumerOffsetChecker, hereis what I get from myJavacode:currentOffset=5;earliestOffset=29; latestOffset=29 'currentOffset'seems to be Ok, 'latestOffset' is correct too, but the 'earliestOffset'? Iwould expect it to be at least '5'? Q4: How could ithappen that the earliestOffset is higher than the currentOffset?My only suspicion isthat maybe messages from the topic were cleaned out due to the retention policy….Any other cases this could have happened?  Thanks a lot!Marina
  

Re: Java API offset operations clarification

[Marina <pp...@yahoo.com.INVALID>]

Sorry, the formatting seems to be all screwed up... I'll try to make it all plain text:

Hi,
I'm trying to use low-level Consumer  Java API to manage offsets manually, with the latest kafka_2.10-0.8.2.1
To verify that the offsets I commit/read form Kafka are correct, I use the kafka.tools.ConsumerOffsetChecker tool.
Here is an example of the output for a topic/consumer group:

./bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group elastic_search_group --zookeeper localhost:2181 --topic my_log_topic
Group           Topic                          Pid Offset          logSize         Lag             Owner
elastic_search_group my_log_topic              0   5               29              24              none

Here is my interpretation of the result:
Offset = 5 --> this is the current offset of my 'elastic_search_group' consumer
logSize = 29 --> this is the Latest offset - the offset of the next message that will come to this topic/partition
Lag = 24  --> 29-5 - how many messages are not yet processed by my 'elastic_search_group' consumer

Q1: is my interpretation correct? And what is the Pid in this case?

Now, I want to get the same information from my Java consumer. Here, I found that I had to use two different APIs:
kafka.javaapi.OffsetRequest to get Earliest and Latest offsets, but kafka.javaapi.OffsetFetchRequest to get the current offset.

To get the Earliest (or Latest) offset I do:

TopicAndPartition topicAndPartition = new TopicAndPartition(myTopic, myPartition);
Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(OffsetRequest.EarliestTime(), 1));
// OR for Latest: requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(OffsetRequest.LatestTime(), 1));
kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(
requestInfo, kafka.api.OffsetRequest.CurrentVersion(), clientName);
OffsetResponse response = simpleConsumer.getOffsetsBefore(request);
long[] offsets = response.offsets(topic, partition);
long myEarliestOffset = offsets[0];
// OR for Latest: long myLatestOffset = offsets[0];


And to get the current offset I have to use a completely different API:

short versionID = 0;
int correlationId = 0;
List<TopicAndPartition> topicPartitionList = new ArrayList<TopicAndPartition>(); 
TopicAndPartition myTopicAndPartition = new TopicAndPartition(myTopic, myPartition);
topicPartitionList.add(myTopicAndPartition);
OffsetFetchRequest offsetFetchReq = new OffsetFetchRequest(
kafkaGroupId, topicPartitionList, versionID, correlationId, kafkaGroupId);
OffsetFetchResponse offsetFetchResponse = simpleConsumer.fetchOffsets(offsetFetchReq);
long
currentOffset = offsetFetchResponse.offsets().get(myTopicAndPartition).offset();


Q2: is it correct? why are there two different APIs to get a very similar information?
Q3: does it matter which versionId and correlationId I am using here? I though  versionId should be 0 for pre-0.8.2.1 kafka, and be 1 for 0.8.2.1 and later - but seems like it works with 0 for 0.8.2.1 as well - see below ?

So, for the the example state of the topic above, and the above output of the ConsumerOffsetChecker, here is what I get from my
Java code:
currentOffset=5; earliestOffset=29; latestOffset=29

'currentOffset' seems to be Ok, 'latestOffset' is correct too, but the 'earliestOffset'? I would expect it to be at least '5'?

Q4: How could it happen that the earliestOffset is higher than the currentOffset?
My only suspicion is that maybe messages from the topic were cleaned out due to the retention policy…. Any other cases this could have happened?

Thanks a lot!
Marina





______________________________