[jira] [Created] (SAMZA-913) CoordinatorStreamSystemConsumer drops messages when they are considered equivalent

["Jake Maes (JIRA)" <ji...@apache.org>]

Jake Maes created SAMZA-913:
-------------------------------

             Summary: CoordinatorStreamSystemConsumer drops messages when they are considered equivalent
                 Key: SAMZA-913
                 URL: https://issues.apache.org/jira/browse/SAMZA-913
             Project: Samza
          Issue Type: Bug
            Reporter: Jake Maes
            Assignee: Jake Maes


When CoordinatorStreamSystemConsumer bootstraps, it adds the messages to a LinkedHashSet ("bootstrappedStreamSet"). The intent seems to be:
1. Messages will be processed in the order they were consumed.
2. Only the latest copy of a message will be stored. 

That second assumption turns out to be false with the current implementation. In Java, Set.add() only adds an element if it doesn't already exist in the Set. Further, CoordinatorStreamMessage.equals() relies on the key set  and values, but not the message offset or timestamp, so the following set of messages could occur:

key1 -> value1  // added to bootstrappedStreamSet
key1 -> value2  // added to bootstrappedStreamSet
key1 -> value1  // duplicate to first message, not added

Thus the final state will be (incorrectly):
key1 -> value2





--
This message was sent by Atlassian JIRA
(v6.3.4#6332)