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了。

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

其实合理的处理方式应该为如果这个时间点前没有值,就返回最早的offset了,对api的使用者就友好多了我们可以自己来实现这个功能。

代码如下:

 

/*

      * 得到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));

           }

            //如果返回的数据长度为0,就获取最早的offset。

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

return getOffset(kafka.api.OffsetRequest

                           . EarliestTime());

           }

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

     }

分享给朋友:
您可能感兴趣的文章:
随机阅读: