[FLINK-23170] Write metadata after materialization

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.15.0
• Component/s: Runtime / State Backends
• Labels: pull-request-available
• Release Note: merged commit 98ec091 into apache:master
• Assignee: Yuan Mei
• Reporter: Roman Khachatryan
• Created: 28/Jun/21 13:22
• Updated: 12/Dec/21 08:16
• Resolved: 12/Dec/21 08:16

Description

Currently, changelog state backend writes state metadata on first state access. It is written to the changelog
On materialization, the changelog can be truncated, so the metadata needs to be written again.

 

Below is a proposed solution using the existing metadtaWritten flag.

An alternative would be to write metadata at the end of the materialized stream.
Yet another approach is to write metadata to a separate file (however, it seems less optimal than at the end of the materialized stream and not so easy as writing again).

There are several questions to answer:

• When to mark the metadata as not written (i.e. reset the metadataWritten flag)?
  • After starting the materialization - so that any subsequent data is preceded by metadata
• When to request the write (i.e. call append)
      At any point (mat. start / mat. end / checkpoint start). It doesn't matter for correctness - see the next points.
  Scheduling append earlier means:
  • including metadata in changelog twice unnecesserily (won't hurt correctness)
  • writing for nothing if materialization fails

Scheduling append later means slowing down the checkpoint
So at materialization end seem to be a better tradeoff.

• What metadata to write?
       Only for data which were changed after materialization started (so the flag is enough)
• Where in changelog to write it to?
      No choice but to the end of the changelog. Because of updating SQN, the metadata will appear at the beginning of the state object returned by persist(sqn) called after materialization completes.
• How to wait for write completion (before completing checkpoint)?
  Once appended, the future returned from persist() call should include it already
   

So to achieve this it's enough to call appendMetadata() for each changed state upon materialization start, or finish, or 1st checkpoint after it.

—
Another related change is to skip writing metadata on recovery (only if state was read from the changelog).
This can be achieved by setting the flag when requesting the state from ChangeLogApplier.
Please create a separate ticket for that if not implementing in this one.

—
Note: with TM-side state ownership, actual log truncation may be delayed after materialization (until all the checkpoints using the log are subsumed). This should not affect the above logic.
 
 

Issue Links

• blocks: FLINK-23559 Randomize periodic materialisation interval in tests
• blocks: FLINK-25143 Add ITCase for Generalized incremental checkpoints
• blocks: FLINK-25144 Test Changelog State Backend manually
• is blocked by: FLINK-21357 Add periodic materialization
• links to: GitHub Pull Request #17462