设为首页 加入收藏

TOP

kafka使用getOffsetsBefore()获取获取offset异常分析(一)
2017-10-10 12:13:18 】 浏览:5087
Tags:kafka 使用 getOffsetsBefore 获取 offset 异常 分析

根据时间戳获取kafka的topic的偏移量,结果获取的偏移量量数据组的长度为0,就会出现如下的数组下标越界的异常,实现的原理是使用了kafka的getOffsetsBefore()方法:

 

Exception in thread "main" java.lang.ArrayIndexOutOfBoundsException : 0

     at co.gridport.kafka.hadoop.KafkaInputFetcher.getOffset(KafkaInputFetcher.java:126)

     at co.gridport.kafka.hadoop.TestKafkaInputFetcher.testGetOffset(TestKafkaInputFetcher.java:68)

     at co.gridport.kafka.hadoop.TestKafkaInputFetcher.main(TestKafkaInputFetcher.java:80)

OffsetResponse(0,Map([barrage_detail,0] -> error: kafka.common.UnknownException offsets: ))

源码如下:

/*

      * 得到partition的offset Finding Starting Offset for Reads

      */

public Long getOffset(Long time) throws IOException {

           TopicAndPartition topicAndPartition = new TopicAndPartition(this.topic , this.partition );

           Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();

requestInfo.put( topicAndPartition, new PartitionOffsetRequestInfo(time, 1));

           kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(

requestInfo, kafka.api.OffsetRequest.CurrentVersion(), this. client_id);

           OffsetResponse response = this. kafka_consumer.getOffsetsBefore( request);

if ( response.hasError()) {

log.error( "Error fetching data Offset Data the Broker. Reason: " + response.errorCode(this.topic, this. partition));

throw new IOException ( "Error fetching kafka Offset by time:" + response.errorCode(this.topic, this. partition));

           }

//         if (response.offsets(this.topic, this.partition).length == 0){

//              return getOffset(kafka.api.OffsetRequest

//                         .EarliestTime());

//         }

return response.offsets( this. topic, this. partition)[0];

     }

返回的response对象会有error: kafka.common.UnknownException offsets如下异常:

OffsetResponse(0,Map([barrage_detail,0] -> error: kafka.common.UnknownException offsets: ))

同时呢,response.hasError()检查不到error。

是什么原因造成了response.offsets(this.topic,this.partition)的返回数组的长度为0呢?

分析了getOffsetsBefore()方法的源码,并做源码了大量的测试,终于重现了这种情况?

1.getOffsetsBefore()的功能以及实现原理:

getOffsetsBefore的功能是返回某个时间点前的maxOffsetNum个offset(时间点指的是kafka日志文件的最后修改时间,offset指的是log文件名中的offset,这个offset是该日志文件的第一条记录的offset,即base offset;maxNumOffsets参数即返回结果的最大个数,如果该参数为2,就返回指定时间点前的2个offset,如果是负数,就报逻辑错误,怎么能声明一个长度为负数的数组呢,呵呵);

根据这个实现原理,所以返回的结果长度为0是合理的,反映的是这个时间点前没有kafka日志这种情况,该情况自然就没有offset了。

说明我们指定的时间参数太早了,正常的时间范围为:最早的时间之后的时间参数都可以有返回值

首页 上一页 1 2 下一页 尾页 1/2/2
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇Scalaz(28)- ST Monad :FP方.. 下一篇Linux下安装scala

最新文章

热门文章

Hot 文章

Python

C 语言

C++基础

大数据基础

linux编程基础

C/C++面试题目