[SAMZA-267] OffsetManager fails if a checkpointed topic isn't in task.inputs

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.6.0
• Fix Version/s: 0.7.0
• Component/s: container
• Assignee: Chris Riccomini
• Reporter: Chris Riccomini
• Created: 15/May/14 21:49
• Updated: 29/May/14 16:01
• Resolved: 29/May/14 16:01

Description

If you run a job with task.inputs=foo, let the job checkpoint, then restart it with task.inputs=bar, the last checkpoint will have foo in it. This will cause the OffsetManager to fail with:

2014-05-15 12:16:28 SamzaContainer [ERROR] Caught exception in process loop.
org.apache.samza.SamzaException: Attempting to reset a stream that doesn't have offset settings SystemStream [system=kafka, stream=foo].
    at org.apache.samza.checkpoint.OffsetManager$$anonfun$getSystemStreamPartitionsToReset$1$$anonfun$apply$5.apply(OffsetManager.scala:305)
    at org.apache.samza.checkpoint.OffsetManager$$anonfun$getSystemStreamPartitionsToReset$1$$anonfun$apply$5.apply(OffsetManager.scala:305)
    at scala.collection.MapLike$class.getOrElse(MapLike.scala:128)
    at scala.collection.AbstractMap.getOrElse(Map.scala:58)
    at org.apache.samza.checkpoint.OffsetManager$$anonfun$getSystemStreamPartitionsToReset$1.apply(OffsetManager.scala:305)
    at org.apache.samza.checkpoint.OffsetManager$$anonfun$getSystemStreamPartitionsToReset$1.apply(OffsetManager.scala:302)
    at scala.collection.TraversableLike$$anonfun$filter$1.apply(TraversableLike.scala:264)
    at scala.collection.Iterator$class.foreach(Iterator.scala:727)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
    at scala.collection.MapLike$DefaultKeySet.foreach(MapLike.scala:174)
    at scala.collection.TraversableLike$class.filter(TraversableLike.scala:263)
    at scala.collection.AbstractTraversable.filter(Traversable.scala:105)
    at org.apache.samza.checkpoint.OffsetManager.getSystemStreamPartitionsToReset(OffsetManager.scala:302)
    at org.apache.samza.checkpoint.OffsetManager.stripResetStreams(OffsetManager.scala:287)
    at org.apache.samza.checkpoint.OffsetManager.start(OffsetManager.scala:165)
    at org.apache.samza.container.SamzaContainer.startOffsetManager(SamzaContainer.scala:558)
    at org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:492)
    at org.apache.samza.container.SamzaContainer$.main(SamzaContainer.scala:82)
    at org.apache.samza.container.SamzaContainer.main(SamzaContainer.scala)

We should just warn in this case, rather than fail the container.