[
https://issues.apache.org/jira/browse/FLINK-27101?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17521426#comment-17521426
]
Yun Tang commented on FLINK-27101:
----------------------------------
Just like the factory of checkpoint store, from my mind, we can still extract
the logic of triggering checkpoint within checkpoint coordinator. Current
default checkpoint triggering policy is start with random delay and then
trigger periodically with same checkpoint options. If we support factory to
create such policy, users could define their policy so that they could trigger
full checkpoint periodically to break the lineage of incremental checkpints. In
the future, some users could also trigger native savepoint periodically with
their customized policy.
Although I did not think about the plan in details, I think this could be
practicable. And it certainlly needs a FLIP for discussion.
> Periodically break the chain of incremental checkpoint
> ------------------------------------------------------
>
> Key: FLINK-27101
> URL: https://issues.apache.org/jira/browse/FLINK-27101
> Project: Flink
> Issue Type: New Feature
> Components: Runtime / Checkpointing
> Reporter: Steven Zhen Wu
> Priority: Major
>
> Incremental checkpoint is almost a must for large-state jobs. It greatly
> reduces the bytes uploaded to DFS per checkpoint. However, there are a few
> implications from incremental checkpoint that are problematic for production
> operations. Will use S3 as an example DFS for the rest of description.
> 1. Because there is no way to deterministically know how far back the
> incremental checkpoint can refer to files uploaded to S3, it is very
> difficult to set S3 bucket/object TTL. In one application, we have observed
> Flink checkpoint referring to files uploaded over 6 months ago. S3 TTL can
> corrupt the Flink checkpoints.
> S3 TTL is important for a few reasons
> - purge orphaned files (like external checkpoints from previous deployments)
> to keep the storage cost in check. This problem can be addressed by
> implementing proper garbage collection (similar to JVM) by traversing the
> retained checkpoints from all jobs and traverse the file references. But that
> is an expensive solution from engineering cost perspective.
> - Security and privacy. E.g., there may be requirement that Flink state can't
> keep the data for more than some duration threshold (hours/days/weeks).
> Application is expected to purge keys to satisfy the requirement. However,
> with incremental checkpoint and how deletion works in RocksDB, it is hard to
> set S3 TTL to purge S3 files. Even though those old S3 files don't contain
> live keys, they may still be referrenced by retained Flink checkpoints.
> 2. Occasionally, corrupted checkpoint files (on S3) are observed. As a
> result, restoring from checkpoint failed. With incremental checkpoint, it
> usually doesn't help to try other older checkpoints, because they may refer
> to the same corrupted file. It is unclear whether the corruption happened
> before or during S3 upload. This risk can be mitigated with periodical
> savepoints.
> It all boils down to periodical full snapshot (checkpoint or savepoint) to
> deterministically break the chain of incremental checkpoints. Search the jira
> history, the behavior that FLINK-23949 [1] trying to fix is actually close to
> what we would need here.
> There are a few options
> 1. Periodically trigger savepoints (via control plane). This is actually not
> a bad practice and might be appealing to some people. The problem is that it
> requires a job deployment to break the chain of incremental checkpoint.
> periodical job deployment may sound hacky. If we make the behavior of full
> checkpoint after a savepoint (fixed in FLINK-23949) configurable, it might be
> an acceptable compromise. The benefit is that no job deployment is required
> after savepoints.
> 2. Build the feature in Flink incremental checkpoint. Periodically (with some
> cron style config) trigger a full checkpoint to break the incremental chain.
> If the full checkpoint failed (due to whatever reason), the following
> checkpoints should attempt full checkpoint as well until one successful full
> checkpoint is completed.
> 3. For the security/privacy requirement, the main thing is to apply
> compaction on the deleted keys. That could probably avoid references to the
> old files. Is there any RocksDB compation can achieve full compaction of
> removing old delete markers. Recent delete markers are fine
> [1] https://issues.apache.org/jira/browse/FLINK-23949
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
