[
https://issues.apache.org/jira/browse/FLINK-27101?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17522243#comment-17522243
]
Piotr Nowojski commented on FLINK-27101:
----------------------------------------
I've chatted offline with [~yunta] a bit about this. I think there are four
options:
# Add a specific configuration property just for this requested feature
# Make checkpoint triggering mechanism pluggable as [~yunta] proposed
# Expose triggering checkpoint via CLI and/or REST API with some parameters to
choose incremental/full checkpoint. We were discussing this before FLINK-6755.
Back then there was no clear motivation behind it, but it should be something
pretty straightforward thing to do
# Not do anything
Both 2. and 3. would require some discussion/small FLIP on the dev mailing list
to decide on the public API. I would be slightly in favour of the option 3., as
IMO it would be easier to implement by us, easier to use by users and more
flexible. For example it would also allow to fully disable triggering automatic
checkpoints in Flink, and let some external system decide when to do it. Also
REST API would be IMO cleaner API compared to the pluggable triggering
mechanism (currently {{CheckpointRequestDecider}} has pretty nasty interface
with quite a bit of lambda functions tightly connecting it to the
{{CheckpointCoordinator}}).
> 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)
