[FLINK-5715] Asynchronous snapshots for heap keyed state backend (BACKPORT)

[StefanRRichter]

Backport of PR #3466 from 1.3-snapshot to 1.2. Introduces asynchronous snapshots for heap keyed state backend.

[StefanRRichter]

CC @StephanEwen

[StephanEwen]

Backport looks good.

Given that for a bugfix release, we should not refactor any implementation (to avoid introducing new issues) this patch must be as "optional" as possible, meaning that it really only adds new files and does not change existing files unless strictly necessary.

It mostly does that, which is very nice. If possible I would

• undo the changes in TimeWindow.java
• undo the change in MemoryStateBackend.java
• add the functions in MathUtil without refactoring the existing function to call the new function

That way we have it as a quasi "add only", modulo the one interface added to the keyed state backend.

[StephanEwen]

To make this work cross platform, always use file.toUri() or new Path(file.toUri()).

[StefanRRichter]

That is true. This similar issue should have been there, because this is basically just copy-paste from the case FILE. I will fix both.

[StefanRRichter]

Thanks for the review, @StephanEwen ! A agree with undoing MemoryStateBackend and MathUtil as proposed, but would argue for keeping the change in TimeWindow. This change fixes skew problems with the old way the hash code was generated (working on timestamps, typically only a small bandwith of bits in the hash code was used by the other approach). This skew can have particular impact in the new, flattened hashmap structure of the async backends.

[StefanRRichter]

After a discussion with @StephanEwen , we decided to follow my proposal. Merging this now.

[StefanRRichter]

Merged in c6a8072.