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

Attach files

Attach Screenshot

Voters

Watch issue

Watchers

Create sub-task

Link

Clone

Update Comment Author

Replace String in Comment

Update Comment Visibility

Delete Comments

XML

Word

Printable

JSON

Details

Type: Bug
Status: Resolved
Priority: Trivial
Resolution: Fixed
Affects Version/s: 0.10.0
Fix Version/s: 0.14.0
Component/s: None
Labels:
None

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.