Spark streaming kafka OffsetOutOfRangeException 异常分析与解决

桃扇骨 2022-06-18 03:19 163阅读 0赞

[Spark streaming kafka OffsetOutOfRangeException 异常分析与解决][Spark streaming kafka OffsetOutOfRangeException]

原文地址：

http://blog.csdn.net/xueba207/article/details/51174818

自从把[Spark][] 从1.3升级到1.6之后，kafka Streaming相关问题频出。最近又遇到了一个。   
job中使用Kafka DirectStream 读取topic中数据，然后做处理。其中有个[测试][Link 1]job，停止了几天，再次启动时爆出了kafka.common.OffsetOutOfRangeException。下文记录下异常分析与解决过程。

## 异常分析 ##

从字面意思上，说是kafka topic的offset越界异常；在job中使用的是Kafka DirectStream，每成功处理一批数据，就把对应的offset更新到zookeeper中；和数组越界异常一样，offset越界应该分为头越界和尾越界，如下图所示。   
![越界示意图][20160417193855850]

*  头部越界： zookeeper中保存的offset在topic中仍然存在的最老message的offset之前时(zk\_offset < earliest\_offset)；
 *  尾部越界： zookeeper中保存的offset在topic中最新message的offset之后时(zk\_offset > last\_offset)

因为代码中采用了[之前文章的方法][Link 2]，因此不可能是尾部越界，因此猜测是头部越界。   
是什么导致头部越界呢？   
考虑到kafka broker配置中修改了message的保持时间为24小时：

> log.retention.hours=24(The minimum age of a log file to be eligible for deletion)
> 
> 因此，应该是kafka 中未被消费的数据被broker清除了，使得zk中的offset落在仍存在的最老message offset的左侧，本来合法的offset变得不非法了。

## 验证猜测 ##

1.  改kafka broker 的retention time 为2分钟   
    配置文件   
    kafka/config/server.properties   
    `log.retention.hours=168 -> log.retention.minutes=2`   
    修改完成后重启kafka。
2.  使用zk shell 命令得到解析器所保存的zk\_offset
3.  停止spark streaming kafka DirectStream job
4.  发送数据到kafka topic，等待一段时间（超过两分钟）
5.  启动streaming job，复现该异常。

通过异常验证可以导致异常的原因为：kafka broker因为log.retention.hours的配置，导致topic中有些数据被清除，而在retention时间范围内streaming job都没有把将要被清除的message消费掉，因此zk中offset落在了earliest\_offset的左侧，引发异常。

## 解决方法 ##

首先想到的方法就是 streaming job要及时消费掉topic中的数据，消费延迟不得大于log.retention.time的配置。   
但是更好的办法是在遇到该问题时，依然能让job正常运行，因此就需要在发现`zk_offset<earliest_offset`时矫正zk\_offset为合法值。   
同样使用[Spark Streaming ‘numRecords must not be negative’问题解决，解决思路][Spark Streaming _numRecords must not be negative]的方法。   
代码：

package com.frey.v1.utils.kafka;
    
    import com.google.common.collect.Lists;
    import com.google.common.collect.Maps;
    import kafka.api.PartitionOffsetRequestInfo;
    import kafka.cluster.Broker;
    import kafka.common.TopicAndPartition;
    import kafka.javaapi.*;
    import kafka.javaapi.consumer.SimpleConsumer;
    
    import java.util.Date;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
    /** * KafkaOffsetTool * * @author FREY * @date 2016/4/11 */
    public class KafkaOffsetTool { 
    
      private static KafkaOffsetTool instance;
      final int TIMEOUT = 100000;
      final int BUFFERSIZE = 64 * 1024;
    
      private KafkaOffsetTool() {
      }
    
      public static synchronized KafkaOffsetTool getInstance() {
        if (instance == null) {
          instance = new KafkaOffsetTool();
        }
        return instance;
      }
    
    public Map<TopicAndPartition, Long> getLastOffset(String brokerList, List<String> topics,
          String groupId) {
    
        Map<TopicAndPartition, Long> topicAndPartitionLongMap = Maps.newHashMap();
    
        Map<TopicAndPartition, Broker> topicAndPartitionBrokerMap =
            KafkaOffsetTool.getInstance().findLeader(brokerList, topics);
    
        for (Map.Entry<TopicAndPartition, Broker> topicAndPartitionBrokerEntry : topicAndPartitionBrokerMap
            .entrySet()) {
          // get leader broker
          Broker leaderBroker = topicAndPartitionBrokerEntry.getValue();
    
          SimpleConsumer simpleConsumer = new SimpleConsumer(leaderBroker.host(), leaderBroker.port(),
              TIMEOUT, BUFFERSIZE, groupId);
    
          long readOffset = getTopicAndPartitionLastOffset(simpleConsumer,
              topicAndPartitionBrokerEntry.getKey(), groupId);
    
          topicAndPartitionLongMap.put(topicAndPartitionBrokerEntry.getKey(), readOffset);
    
        }
    
        return topicAndPartitionLongMap;
    
      }
    
      /** * * @param brokerList * @param topics * @param groupId * @return */
      public Map<TopicAndPartition, Long> getEarliestOffset(String brokerList, List<String> topics,
          String groupId) {
    
        Map<TopicAndPartition, Long> topicAndPartitionLongMap = Maps.newHashMap();
    
        Map<TopicAndPartition, Broker> topicAndPartitionBrokerMap =
            KafkaOffsetTool.getInstance().findLeader(brokerList, topics);
    
        for (Map.Entry<TopicAndPartition, Broker> topicAndPartitionBrokerEntry : topicAndPartitionBrokerMap
            .entrySet()) {
          // get leader broker
          Broker leaderBroker = topicAndPartitionBrokerEntry.getValue();
    
          SimpleConsumer simpleConsumer = new SimpleConsumer(leaderBroker.host(), leaderBroker.port(),
              TIMEOUT, BUFFERSIZE, groupId);
    
          long readOffset = getTopicAndPartitionEarliestOffset(simpleConsumer,
              topicAndPartitionBrokerEntry.getKey(), groupId);
    
          topicAndPartitionLongMap.put(topicAndPartitionBrokerEntry.getKey(), readOffset);
    
        }
    
        return topicAndPartitionLongMap;
    
      }
    
      /** * 得到所有的 TopicAndPartition * * @param brokerList * @param topics * @return topicAndPartitions */
      private Map<TopicAndPartition, Broker> findLeader(String brokerList, List<String> topics) {
        // get broker's url array
        String[] brokerUrlArray = getBorkerUrlFromBrokerList(brokerList);
        // get broker's port map
        Map<String, Integer> brokerPortMap = getPortFromBrokerList(brokerList);
    
        // create array list of TopicAndPartition
        Map<TopicAndPartition, Broker> topicAndPartitionBrokerMap = Maps.newHashMap();
    
        for (String broker : brokerUrlArray) {
    
          SimpleConsumer consumer = null;
          try {
            // new instance of simple Consumer
            consumer = new SimpleConsumer(broker, brokerPortMap.get(broker), TIMEOUT, BUFFERSIZE,
                "leaderLookup" + new Date().getTime());
    
            TopicMetadataRequest req = new TopicMetadataRequest(topics);
    
            TopicMetadataResponse resp = consumer.send(req);
    
            List<TopicMetadata> metaData = resp.topicsMetadata();
    
            for (TopicMetadata item : metaData) {
              for (PartitionMetadata part : item.partitionsMetadata()) {
                TopicAndPartition topicAndPartition =
                    new TopicAndPartition(item.topic(), part.partitionId());
                topicAndPartitionBrokerMap.put(topicAndPartition, part.leader());
              }
            }
          } catch (Exception e) {
            e.printStackTrace();
          } finally {
            if (consumer != null)
              consumer.close();
          }
        }
        return topicAndPartitionBrokerMap;
      }
    
      /** * get last offset * @param consumer * @param topicAndPartition * @param clientName * @return */
      private long getTopicAndPartitionLastOffset(SimpleConsumer consumer,
          TopicAndPartition topicAndPartition, String clientName) {
        Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo =
            new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
    
        requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(
            kafka.api.OffsetRequest.LatestTime(), 1));
    
        OffsetRequest request = new OffsetRequest(
            requestInfo, kafka.api.OffsetRequest.CurrentVersion(),
            clientName);
    
        OffsetResponse response = consumer.getOffsetsBefore(request);
    
        if (response.hasError()) {
          System.out
              .println("Error fetching data Offset Data the Broker. Reason: "
                  + response.errorCode(topicAndPartition.topic(), topicAndPartition.partition()));
          return 0;
        }
        long[] offsets = response.offsets(topicAndPartition.topic(), topicAndPartition.partition());
        return offsets[0];
      }
    
      /** * get earliest offset * @param consumer * @param topicAndPartition * @param clientName * @return */
      private long getTopicAndPartitionEarliestOffset(SimpleConsumer consumer,
          TopicAndPartition topicAndPartition, String clientName) {
        Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo =
            new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
    
        requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(
            kafka.api.OffsetRequest.EarliestTime(), 1));
    
        OffsetRequest request = new OffsetRequest(
            requestInfo, kafka.api.OffsetRequest.CurrentVersion(),
            clientName);
    
        OffsetResponse response = consumer.getOffsetsBefore(request);
    
        if (response.hasError()) {
          System.out
              .println("Error fetching data Offset Data the Broker. Reason: "
                  + response.errorCode(topicAndPartition.topic(), topicAndPartition.partition()));
          return 0;
        }
        long[] offsets = response.offsets(topicAndPartition.topic(), topicAndPartition.partition());
        return offsets[0];
      }
      /** * 得到所有的broker url * * @param brokerlist * @return */
      private String[] getBorkerUrlFromBrokerList(String brokerlist) {
        String[] brokers = brokerlist.split(",");
        for (int i = 0; i < brokers.length; i++) {
          brokers[i] = brokers[i].split(":")[0];
        }
        return brokers;
      }
    
      /** * 得到broker url 与 其port 的映射关系 * * @param brokerlist * @return */
      private Map<String, Integer> getPortFromBrokerList(String brokerlist) {
        Map<String, Integer> map = new HashMap<String, Integer>();
        String[] brokers = brokerlist.split(",");
        for (String item : brokers) {
          String[] itemArr = item.split(":");
          if (itemArr.length > 1) {
            map.put(itemArr[0], Integer.parseInt(itemArr[1]));
          }
        }
        return map;
      }
    
      public static void main(String[] args) {
        List<String> topics = Lists.newArrayList();
        topics.add("my_topic");
    // topics.add("bugfix");
        Map<TopicAndPartition, Long> topicAndPartitionLongMap =
            KafkaOffsetTool.getInstance().getEarliestOffset("broker1:9092,broker2:9092", topics,
                "com.frey.group");
    
        for (Map.Entry<TopicAndPartition, Long> entry : topicAndPartitionLongMap.entrySet()) {
         System.out.println(entry.getKey().topic() + "-"+ entry.getKey().partition() + ":" + entry.getValue());
        }
      }
    }
    
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
       40
       41
       42
       43
       44
       45
       46
       47
       48
       49
       50
       51
       52
       53
       54
       55
       56
       57
       58
       59
       60
       61
       62
       63
       64
       65
       66
       67
       68
       69
       70
       71
       72
       73
       74
       75
       76
       77
       78
       79
       80
       81
       82
       83
       84
       85
       86
       87
       88
       89
       90
       91
       92
       93
       94
       95
       96
       97
       98
       99
       100
       101
       102
       103
       104
       105
       106
       107
       108
       109
       110
       111
       112
       113
       114
       115
       116
       117
       118
       119
       120
       121
       122
       123
       124
       125
       126
       127
       128
       129
       130
       131
       132
       133
       134
       135
       136
       137
       138
       139
       140
       141
       142
       143
       144
       145
       146
       147
       148
       149
       150
       151
       152
       153
       154
       155
       156
       157
       158
       159
       160
       161
       162
       163
       164
       165
       166
       167
       168
       169
       170
       171
       172
       173
       174
       175
       176
       177
       178
       179
       180
       181
       182
       183
       184
       185
       186
       187
       188
       189
       190
       191
       192
       193
       194
       195
       196
       197
       198
       199
       200
       201
       202
       203
       204
       205
       206
       207
       208
       209
       210
       211
       212
       213
       214
       215
       216
       217
       218
       219
       220
       221
       222
       223
       224
       225
       226
       227
       228
       229
       230
       231
       232
       233
       234
       235
       236
       237
       238
       239
       240
       241
       242
       243
       244
       245
       246
       247
       248
       249
       250
       251
       252
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
       40
       41
       42
       43
       44
       45
       46
       47
       48
       49
       50
       51
       52
       53
       54
       55
       56
       57
       58
       59
       60
       61
       62
       63
       64
       65
       66
       67
       68
       69
       70
       71
       72
       73
       74
       75
       76
       77
       78
       79
       80
       81
       82
       83
       84
       85
       86
       87
       88
       89
       90
       91
       92
       93
       94
       95
       96
       97
       98
       99
       100
       101
       102
       103
       104
       105
       106
       107
       108
       109
       110
       111
       112
       113
       114
       115
       116
       117
       118
       119
       120
       121
       122
       123
       124
       125
       126
       127
       128
       129
       130
       131
       132
       133
       134
       135
       136
       137
       138
       139
       140
       141
       142
       143
       144
       145
       146
       147
       148
       149
       150
       151
       152
       153
       154
       155
       156
       157
       158
       159
       160
       161
       162
       163
       164
       165
       166
       167
       168
       169
       170
       171
       172
       173
       174
       175
       176
       177
       178
       179
       180
       181
       182
       183
       184
       185
       186
       187
       188
       189
       190
       191
       192
       193
       194
       195
       196
       197
       198
       199
       200
       201
       202
       203
       204
       205
       206
       207
       208
       209
       210
       211
       212
       213
       214
       215
       216
       217
       218
       219
       220
       221
       222
       223
       224
       225
       226
       227
       228
       229
       230
       231
       232
       233
       234
       235
       236
       237
       238
       239
       240
       241
       242
       243
       244
       245
       246
       247
       248
       249
       250
       251
       252

矫正offset核心代码：

/** 以下 矫正 offset */
    
        // lastest offsets
        Map<TopicAndPartition, Long> lastestTopicAndPartitionLongMap =
            KafkaOffsetTool.getInstance().getLastOffset(kafkaParams.get("metadata.broker.list"),
                Lists.newArrayList(topicsSet), kafkaParams.get(Constants.KAFKA_CONSUMER_GROUP_ID));
    
        // earliest offsets
        Map<TopicAndPartition, Long> earliestTopicAndPartitionLongMap =
            KafkaOffsetTool.getInstance().getEarliestOffset(kafkaParams.get("metadata.broker.list"),
                Lists.newArrayList(topicsSet), kafkaParams.get(Constants.KAFKA_CONSUMER_GROUP_ID));
    
    
        for (Map.Entry<TopicAndPartition, Long> topicAndPartitionLongEntry : fromOffsets.entrySet()) {
    
          long zkOffset = topicAndPartitionLongEntry.getValue();
          long lastestOffset = lastestTopicAndPartitionLongMap.get(topicAndPartitionLongEntry.getKey());
          long earliestOffset = earliestTopicAndPartitionLongMap.get(topicAndPartitionLongEntry.getKey());
          // zkoffset 不在可用message offset区间内
          if (zkOffset > lastestOffset || zkOffset < earliestOffset) {
            // set offset = earliestOffset
            logger.warn("矫正offset: " + zkOffset +" -> "+ earliestOffset);
            topicAndPartitionLongEntry.setValue(earliestOffset);
          }
        }
        /** 以上 矫正 offset */

[Spark streaming kafka OffsetOutOfRangeException]: http://blog.csdn.net/xueba207/article/details/51174818
[Spark]: http://lib.csdn.net/base/spark
[Link 1]: http://lib.csdn.net/base/softwaretest
[20160417193855850]: /images/20220618/ddea4d1948bd4743895df1329c143c6e.png
[Link 2]: http://blog.csdn.net/xueba207/article/details/51135423
[Spark Streaming _numRecords must not be negative]: http://blog.csdn.net/xueba207/article/details/51135423#t6