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
