Kafka Java API偏移操作澄清

Question

我正在尝试使用最低级别的Consumer Java API来手动管理偏移,使用最新的kafka_2.10-0.8.2.1.要验证我从Kafka提交/读取的偏移量是否正确,我使用kafka.tools.ConsumerOffsetChecker工具.

以下是我的主题/使用者组的输出示例:

./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

  以下是我对结果的解释:

Offset = 5 - >这是我'elastic_search_group'消费者的当前偏移量

logSize = 29 - >这是最新偏移量 - 将来到此主题/分区的下一条消息的偏移量

滞后= 24 - > 29-5 - 我的'elastic_search_group'消费者尚未处理多少消息

Pid - 分区ID

Q1:这是对的吗？

现在,我想从我的Java消费者那里获得相同的信息.在这里,我发现我必须使用两种不同的API:

kafka.javaapi.OffsetRequest获得最早和最新的抵消,但kafka.javaapi.OffsetFetchRequest获取当前偏移量.

要获得最早(或最新)的偏移我做:

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];

要获得当前的偏移量,我必须使用完全不同的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:这是对的吗？为什么有两种不同的API来获取非常相似的信息？

问题3:我在这里使用哪个versionId和correlationId有关系吗？虽然versionId对于0.8.2.1之前的kafka应该是0,对于0.8.2.1及更高版本应该是1 - 但是看起来它也适用于0而0.8.2.1 - 见下文？

因此,对于上述主题的示例状态,以及ConsumerOffsetChecker的上述输出,以下是我从Java代码中获得的内容:

currentOffset = 5; earliestOffset = 29; latestOffset = 29

'currentOffset'似乎没问题,'latestOffset'也是正确的,但'earliestOffset'？我希望它至少是'5'？

问题4:如果最早的偏移量高于currentOffset,怎么可能发生？我唯一的怀疑是,由于保留政策,可能会清除主题中的消息.... 这可能发生的任何其他情况？

Answer 1

我正在寻找在分区中找到滞后的方法.这涉及到您采取的相同步骤.到目前为止,无论我学到什么,我都可以给你答案.

1. logSize直接指向该特定分区中已累积的消息数.或者,它指定该分区中消息的最大偏移量.偏移量是上次成功消耗的消息的偏移量.所以滞后只是Log size和Offset之间的差异.
2. 是的,这是对的.到目前为止,这是找到当前偏移和最早或最新偏移的唯一两种方法
3. 我不知道为什么需要指定versionId.您可以使用kafka.api.OffsetRequest.CurrentVersion()获取versionId.因此可以避免硬编码.您可以安全地将correlationId假设为0.

4. 这很奇怪.当我使用EarliestTime()时,即使我当前的偏移量进一步发展,我也将最早的偏移量设为0.这意味着它是分区的开始.因此,当某些消息在将来某个时间到期时,这个最早的偏移将是一些非零数字.现在,如果由于保留策略而清除了消息,则应该更改延迟.我不确定这种行为.一种可靠的方法是,在注意到这样的读取并检查其日志之后运行消费者.它应该显示这样的线条.

   2015-06-09 18:49:15 :: DEBUG :: PartitionTopicInfo:52 :: reset消耗请求偏移量:2:fetched offset = 405952:消耗偏移= 335372到335372 2015-06-09 18:49:15: :DEBUG :: PartitionTopicInfo:52 :: reset消耗请求的偏移量:2:fetched offset = 405952:消耗的偏移量= 335373到335373

请注意,在上面的日志行中,获取的偏移量保持不变,消耗的偏移量也在增加.最后它会结束

2015-06-09 18:49:16 :: DEBUG :: PartitionTopicInfo:52 :: reset消耗请求偏移量:2:fetched offset = 405952:消耗偏移= 405952到405952

那么这意味着由于日志保留策略从335372偏移到405952已过期