[KAFKA-4205] NullPointerException in fetchOffsetsBefore

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.9.0.1
• Fix Version/s: 0.10.1.1
• Component/s: core
• Labels: reliability
• Assignee: Anton Karamanov
• Reporter: Andrew Grasso
• Created: 22/Sep/16 15:44
• Updated: 04/Dec/16 19:35
• Resolved: 04/Dec/16 19:35

Description

We recently observed the following error in brokers running 0.9.0.1:

A client saw an Unkown error code in response to an offset request for TOPICX, partition 0

The server logs look like:

[2016-09-21 21:26:07,143] INFO Scheduling log segment 527235760 for log TOPICX-0 for deletion. (kafka.log.Log)
[2016-09-21 21:26:07,144] ERROR [KafkaApi-13] Error while responding to offset request (kafka.server.KafkaApis)
java.lang.NullPointerException
        at kafka.server.KafkaApis.fetchOffsetsBefore(KafkaApis.scala:513)
        at kafka.server.KafkaApis.fetchOffsets(KafkaApis.scala:501)
        at kafka.server.KafkaApis$$anonfun$18.apply(KafkaApis.scala:461)
        at kafka.server.KafkaApis$$anonfun$18.apply(KafkaApis.scala:452)
        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
        at scala.collection.immutable.Map$Map1.foreach(Map.scala:109)
        at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
        at scala.collection.AbstractTraversable.map(Traversable.scala:105)
        at kafka.server.KafkaApis.handleOffsetRequest(KafkaApis.scala:452)
        at kafka.server.KafkaApis.handle(KafkaApis.scala:70)
        at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:60)
        at java.lang.Thread.run(Thread.java:745)
[2016-09-21 21:27:07,143] INFO Deleting segment 527235760 from log TOPICX-0. (kafka.log.Log)
[2016-09-21 21:27:07,263] INFO Deleting index /path/to/kafka/data/TOPICX-0/00000000000527235760.index.deleted (kafka.log.OffsetIndex)

I suspect a race condition between Log.deleteSegment (which takes a lock on the log) and KafkaApis.fetchOffsetsBefore, which does not take any lock. In particular, line 513 in KafkaApis looks like:

KafkaApis.scala

510  private def fetchOffsetsBefore(log: Log, timestamp: Long, maxNumOffsets: Int): Seq[Long] = {
511    val segsArray = log.logSegments.toArray
512    var offsetTimeArray: Array[(Long, Long)] = null
513    val lastSegmentHasSize = segsArray.last.size > 0;

Issue Links

• links to: GitHub Pull Request #2204