[SAMZA-913] CoordinatorStreamSystemConsumer drops messages when they are considered equivalent

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 0.10.1
• Assignee: Jake Maes
• Reporter: Jake Maes
• Created: 24/Mar/16 21:29
• Updated: 25/Mar/16 20:26
• Resolved: 25/Mar/16 20:26

Description

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

Issue Links

• blocks: SAMZA-906 Host Affinity - Minimize task reassignment when container count changes