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

["Chris Riccomini (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SAMZA-267?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14011579#comment-14011579 ] 

Chris Riccomini commented on SAMZA-267:
---------------------------------------

We need to decide if we want to keep outdated stream offsets in the checkpoint. If we keep the outdated stream offsets in the checkpoint, it means that developers can switch task.inputs back later, and resume where they left off with the old stream. If we strip all outdated (no longer in task.inputs) streams out of the checkpoint, then once a new checkpoint is written, switching back to a previously-consumed stream will result in using the samza.offset.default setting.

I think there are use cases for both behaviors, but I don't want to add yet-another-config to control this. The easiest fix is to just strip out all outdated streams from systemStreamPartitions and lastProcessedOffsets. This would result in stripping all outdated streams from future checkpoints.

The other way to handle this is to fix each individual breakage. There are two places in the OffsetManager where breakage is introduced when input streams are switched.

# getSystemStreamPartitionsToReset
# loadDefaults

For getSystemStreamPartitionsToReset, we could log and ignore outdated streams. For loadDefaults, we could just use the global default (OffsetType.UPCOMING).

> OffsetManager fails if a checkpointed topic isn't in task.inputs
> ----------------------------------------------------------------
>
>                 Key: SAMZA-267
>                 URL: https://issues.apache.org/jira/browse/SAMZA-267
>             Project: Samza
>          Issue Type: Bug
>          Components: container
>    Affects Versions: 0.6.0
>            Reporter: Chris Riccomini
>             Fix For: 0.7.0
>
>
> 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:
> {noformat}
> 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)
> {noformat}
> We should just warn in this case, rather than fail the container.



--
This message was sent by Atlassian JIRA
(v6.2#6252)