[jira] [Commented] (SAMZA-141) TaskInstance.offsets fails to update

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

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

Chris Riccomini commented on SAMZA-141:
---------------------------------------

This is a pretty major issue. It totally breaks checkpointing, since half the time the KafkaCheckpointManager grabs the first offset in the map for each SystemStream, which may or may not be the SSP that's been updated.

> TaskInstance.offsets fails to update
> ------------------------------------
>
>                 Key: SAMZA-141
>                 URL: https://issues.apache.org/jira/browse/SAMZA-141
>             Project: Samza
>          Issue Type: Bug
>          Components: container
>    Affects Versions: 0.6.0
>            Reporter: Chris Riccomini
>            Assignee: Chris Riccomini
>
> TaskInstance has a variable called offsets, which manages offsets for each input stream. The variable is typed to SystemStream -> String. We use SystemString when intracting with it in all places, EXCEPT in the TaskInstance.process method, where we use a SystemStreamPartition as the key. This doesn't work, since a SystemStreamPartition and SystemStream with the same system and stream name are actually NOT equal, since the equality checks do a class check, and the classes don't match.
> As a result, you end up with something like this:
> {code}
>     var offsets = Map[SystemStream, String]()
>     offsets += new SystemStream("foo", "bar") -> "123"
>     offsets += new SystemStreamPartition("foo", "bar", new Partition(0)) -> "321"
>     System.err.println(offsets)
> {code}
> Which prints:
> {noformat}
> Map(SystemStream [system=foo, stream=bar] -> 123, SystemStreamPartition [partition=Partition [partition=0], system=foo, stream=bar] -> 321)
> {noformat}



--
This message was sent by Atlassian JIRA
(v6.1.5#6160)