kafka消费堆积时如何减少消息丢失

使用kafka(0.8.2.1)高级API消费消息时，有时会因各种原因，导致消息堆积。如果请求offset对应消息已过期，则会抛出下面异常：

Current offset 789380 for partition [test,3] out of range; reset offset to 799380"

抛出该异常的同时，会把该topic和分区下次请求的offset重置为一个新的值，此时就发生了消息丢失。

那么，如果减少数据丢失呢？

注意异常后重置offset的值。先看处理offset超出有效范围的代码。

// handle a partition whose offset is out of range and return a new fetch offset
def handleOffsetOutOfRange(topicAndPartition: TopicAndPartition): Long = {
  var startTimestamp : Long = 0
config.autoOffsetReset match {
    case OffsetRequest.SmallestTimeString => startTimestamp = OffsetRequest.EarliestTime
case OffsetRequest.LargestTimeString => startTimestamp = OffsetRequest.LatestTime
case _ => startTimestamp = OffsetRequest.LatestTime
}
  val newOffset = simpleConsumer.earliestOrLatestOffset(topicAndPartition, startTimestamp, Request.OrdinaryConsumerId)
  val pti = partitionMap(topicAndPartition)
  pti.resetFetchOffset(newOffset)
  pti.resetConsumeOffset(newOffset)
  newOffset
}

通过代码可以清楚看到，这个重置的offet值和auto.offset.reset配置有关。auto.offset.reset值有earliest和latest。当消费客户端启动时

earliest 当分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费。
latest 当分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据。

当消费客户端在消费过程中，如果有消息过期，earliest会将offset重置为kafka里最早消息，latest则会将offset重置为最新消息。因此，把auto.offset.reset设置为earliest， 可以在消费堆积的情况下，减少数据丢失。注意，auto.offset.reset默认是latest。