-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/45339/
-----------------------------------------------------------
(Updated March 25, 2016, 6:21 p.m.)
Review request for samza.
Changes
-------
Copying summary from the JIRA
Repository: samza
Description (updated)
-------
SAMZA-913 CoordinatorStreamSystemConsumer drops messages when they are
considered equivalent
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
Diffs
-----
samza-core/src/main/java/org/apache/samza/coordinator/stream/CoordinatorStreamSystemConsumer.java
e1a7626de9ca78ffbffeab65a69605e748ab4479
samza-core/src/test/java/org/apache/samza/coordinator/stream/TestCoordinatorStreamSystemConsumer.java
0e73e18bd55e343e1a5122be7e8f3c666b797dc5
Diff: https://reviews.apache.org/r/45339/diff/
Testing
-------
Added a unit test which fails before the change and passes after.
Ran check-all.sh
This patch fixed my test job for SAMZA-906
Thanks,
Jake Maes
