[SAMZA-1480] TaskStorageManager improperly initializes changelog consumer position when restoring a store from disk

• Type: Bug
• Status: Resolved
• Priority: Trivial
• Resolution: Fixed
• Affects Version/s: 0.10.0
• Fix Version/s: 0.14.0
• Assignee: Jake Maes
• Reporter: Jake Maes
• Created: 01/Nov/17 20:12
• Updated: 09/Nov/17 19:55
• Resolved: 09/Nov/17 19:55

Description

For the Host Affinity state restore, an OFFSET file is written to disk on each commit. This offset file contains the most recently written changelog event which is also reflected in the on-disk state. When the container is restarted, it restores the on-disk store and then replays the changelog from the offset recorded in the OFFSET file in order to restore any changelog events that were produced when the job ran on a different host.
http://samza.apache.org/learn/documentation/0.13/yarn/yarn-host-affinity.html

When TaskStorageManager initializes the consumer, it uses the offset from the OFFSET file, which is already reflected in the state.

Instead, it should use the SystemAdmin.getOffsetsAfter() method to get the next offset to consume. This will avoid the replay of 1 extra message for state restore.

It should then use SystemAdmin.offsetComparator() to use the larger of the next offset (calculated above) and the oldest offset (according to the metadata). This is necessary for changelogs configured with TTL retention rather than infinite retention where the offset from the OFFSET file may no longer be valid.

Issue Links

• links to: GitHub Pull Request #350