Hi, Thanks for updating the FLIP Dawid
There seems to be a consensus in the discussion, however, I couldn't find stop-with-savepoint in the document. A few minor things: - maybe include "checkpoint" in mode names, i.e. --no-claim-checkpoint? - add an explicit option to preserve the current behavior (no claim and no duplicate)? And I still think it would be nice to list object stores which support duplicate operation. Regards, Roman On Fri, Nov 26, 2021 at 10:37 AM Konstantin Knauf <kna...@apache.org> wrote: > > Hi Dawid, > > sounds good, specifically 2., too. > > Best, > > Konstantin > > On Fri, Nov 26, 2021 at 9:25 AM Dawid Wysakowicz <dwysakow...@apache.org> > wrote: > > > Hi all, > > > > I updated the FLIP with a few clarifications: > > > > 1. I added a description how would we trigger a "full snapshot" in the > > changelog state backend > > - (We would go for this option in the 1st version). Trigger a > > snapshot of the base state backend in the 1st checkpoint, which > > induces > > materializing the changelog. In this approach we could duplicate SST > > files, > > but we would not duplicate the diff files. > > - Add a hook for logic for computing which task should duplicate the > > diff files. We would have to do a pass over all states after the state > > assignment in StateAssignmentOperation > > 2. I clarified that the "no-claim" mode requires a > > completed/successful checkpoint before we can remove the one we are > > restoring from. Also added a note that we can assume a checkpoint is > > completed if it is confirmed by Flink's API for checkpointing stats or by > > checking an entry in HA services. A checkpoint can not be assumed > > completed > > by just looking at the checkpoint files. > > > > I suggest going on with the proposal for "no-claim" as suggested so far, > > as it is easier to understand by users. They can reliably tell when they > > can expect the checkpoint to be deletable. If we see that the time to take > > the 1st checkpoint becomes a problem we can extend the set of restore > > methods and e.g. add a "claim-temporarily" method. > > > > I hope we can reach a consensus and start a vote, some time early next > > week. > > > > Best, > > > > Dawid > > > > On 23/11/2021 22:39, Roman Khachatryan wrote: > > > > I also referred to the "no-claim" mode and I still think neither of them > > works in that mode, as you'd have to keep lineage of checkpoints externally > > to be able delete any checkpoint. > > > > I think the lineage is needed in all approaches with arbitrary > > histories; the difference is whether a running Flink is required or > > not. Is that what you mean? > > (If not, could you please explain how the scenario you mentioned above > > with multiple jobs branching from the same checkpoint is handled?) > > > > > > BTW, the state key for RocksDB is actually: backend UID + key group range + > > SST file name, so the key would be different (the key group range is > > different for two tasks) and we would've two separate counters for the same > > file. > > > > You're right. But there is also a collision between old and new entries. > > > > > > To be on the same page here. It is not a problem so far in RocksDB, because > > we do not reuse any shared files in case of rescaling. > > > > As I mentioned above, collision happens not only because of rescaling; > > and AFAIK, there are some ideas to reuse files on rescaling (probably > > Yuan could clarify). Anyways, I think it makes sense to not bake in > > this assumption unless it's hard to implement (or at least state it > > explicitly in FLIP). > > > > > > It is not suggested as an optimization. It is suggested as a must for state > > backends that need it. I did not elaborate on it, because it could affected > > only the changelog state backend at the moment, which I don't have much > > insights. I agree it might make sense to look a bit how we could force full > > snapshots in the changelog state backend. I will spend some extra time on > > that. > > > > I see. For the Changelog state backend, the easiest way would be to > > obtain a full snapshot from the underlying backend in snapshot(), > > ignoring all non-materialized changes. This will effectively > > materialize all the changes, so only new non-materialized state will > > be used in subsequent checkpoints. > > > > > > Only the task that gets assigned [1,16] would be responsible for > > duplicating files of the old range [1, 64]. > > > > Wouldn't it be likely that the same TM will be responsible for [1, 64] > > "windowState", [1, 64] "timerState", and so on, for all operators in > > the chain, and probably other chains? (that what I mean by skew) > > If we want to address this, preserving handle immutability then we'll > > probably have to rebuild the whole task state snapshot. > > (depending on how we approach RocksDB re-uploading, it might not be > > relevant though) > > > > > > Regards, > > Roman > > > > > > On Tue, Nov 23, 2021 at 4:06 PM Dawid Wysakowicz <dwysakow...@apache.org> > > <dwysakow...@apache.org> wrote: > > > > I think I know where the confusion comes from regarding arbitrarily > > recovery histories: Both my counter-proposals were for "no-claim" > > mode; I didn't mean to replace "claim" mode with them. > > However, as Yun pointed out, it's impossible to guarantee that all the > > files will be compacted in a finite number of checkpoints; so let's > > withdraw those proposals. > > > > I also referred to the "no-claim" mode and I still think neither of them > > works in that mode, as you'd have to keep lineage of checkpoints externally > > to be able delete any checkpoint. > > > > Let's consider a job running with DoP=1; it created checkpoint C1 with > > a single file F1 and then stopped. > > We start a new job from C1 in no-claim mode with DoP=2; so two tasks > > will receive the same file F1. > > > > To be on the same page here. It is not a problem so far in RocksDB, because > > we do not reuse any shared files in case of rescaling. If we want to change > > how rescaling in RocksDB works then yes, we would have to consider how we > > want to make sure we copy/duplicate just once. However we would have to > > first change a crucial thing about regular incremental checkpoints and > > reorganize the SharedStateRegistry along the way. > > > > BTW, the state key for RocksDB is actually: backend UID + key group range + > > SST file name, so the key would be different (the key group range is > > different for two tasks) and we would've two separate counters for the same > > file. > > > > Of course, correct me if I am wrong in the two paragraphs above. > > > > Re-upload from one task (proposed in FLIP as optimization) > > > > It is not suggested as an optimization. It is suggested as a must for state > > backends that need it. I did not elaborate on it, because it could affected > > only the changelog state backend at the moment, which I don't have much > > insights. I agree it might make sense to look a bit how we could force full > > snapshots in the changelog state backend. I will spend some extra time on > > that. > > > > Lastly I might be wrong, but I think the KeyedStateHandle#getIntersection > > is a good candidate to distribute the task of duplicating shared files > > pretty evenly. The idea was that we could mark specially the handles that > > are assigned the "start of the old key group range". Therefore if a file > > belonged to a handle responsible for a key group range: [1,64], which is > > later on split into [1, 16], [17, 32], [33, 48]. [49, 64]. Only the task > > that gets assigned [1,16] would be responsible for duplicating files of the > > old range [1, 64]. > > > > Best, > > > > Dawid > > > > On 23/11/2021 14:27, Khachatryan Roman wrote: > > > > Thanks Dawid, Yun and Piotr, > > > > I think I know where the confusion comes from regarding arbitrarily > > recovery histories: Both my counter-proposals were for "no-claim" > > mode; I didn't mean to replace "claim" mode with them. > > However, as Yun pointed out, it's impossible to guarantee that all the > > files will be compacted in a finite number of checkpoints; so let's > > withdraw those proposals. > > > > And as there are no other alternatives left, the changes to > > SharedStateRegistry or State Backends are not a decisive factor > > anymore. > > > > However, it probably still makes sense to clarify the details of how > > re-upload will work in case of rescaling. > > > > Let's consider a job running with DoP=1; it created checkpoint C1 with > > a single file F1 and then stopped. > > We start a new job from C1 in no-claim mode with DoP=2; so two tasks > > will receive the same file F1. > > > > Let's say both tasks will re-use F1, so it needs to be re-uploaded. > > Now, we have a choice: > > 1. Re-upload from both tasks > > For RocksDB, the state key is: backend UID + SST file name. Both are > > the same for two tasks, so the key will be the same. > > Currently, SharedStateRegistry will reject both as duplicates. > > > > We can't just replace (to not lose one of the files), so we have to > > use random keys. > > However, when we further downscale: > > - we'll have a conflict on recovery (multiple SST files with the same name) > > - we'll re-upload the same file multiple times unnecessarily > > So we have to de-duplicate state on recovery - ideally before sending > > state snapshots to tasks. > > > > 2. Re-upload from one task (proposed in FLIP as optimization) > > Both tasks must learn the new key. Otherwise, the snapshot of the > > not-reuploading task will refer to a non-existing entry. > > We can either re-use the old key (and allow replacement in > > SharedStateRegistry); or generate the key on JM before sending task > > state snapshots. > > > > > > P.S.: > > > > 2. Instead of forcing re-upload, can we "inverse control" in no-claim mode? > > This is effectively what we have right now, but with an extra (Async?) > > > > Right now, there is absolutely no way to find out when the shared > > state can be deleted; it can't be inferred from which checkpoints are > > subsumed, and which are not, as future checkpoints might still be > > using that state. > > > > Regards, > > Roman > > > > > > > > On Tue, Nov 23, 2021 at 1:37 PM Piotr Nowojski <pnowoj...@apache.org> > > <pnowoj...@apache.org> wrote: > > > > Hi, > > > > I'm not entirely sure if I fully understand the raised concerns here. So > > let me maybe step back in the discussion a bit and address the original > > points from Roman. > > > > 2. Instead of forcing re-upload, can we "inverse control" in no-claim > > > > mode? > > > > I second the concerns from Dawid. This is effectively what we have right > > now, but with an extra (Async?) API call. It's not conceptually simple, > > it's hard to explain to the users, it might take actually forever to > > release the artefacts. Furthermore I don't think the implementation would > > be trivial. > > > > On the other hand the current proposal of having (a) `--claim` and (b) > > `--no-claim` mode are conceptually very simple. (a) being perfectly > > efficient, without any overheads. If you have concerns that (b) will cause > > some overheads, slower first checkpoint etc, keep in mind that the user can > > always pick option (a). Starting a new job from an existing > > savepoint/externalised checkpoint in general shouldn't be time critical, so > > users can always even manually copy the files and still use option (a), or > > just be fine accepting the price of a slower first checkpoint. For other > > use cases - restarting the same job after a downtime - (b) sounds to me to > > be an acceptable option. > > > > I would also like to point out that the "force full snapshot"/"do not use > > previous artefacts" option we will need either way for the incremental > > intermediate savepoints (subject of a next FLIP). From this perspective, we > > are getting the "--no-claim" option basically for free. > > > > 3. Alternatively, re-upload not necessarily on 1st checkpoint, but after > > > > a configured number of checkpoints? > > > > I don't see a reason why we couldn't provide an option like that at some > > point in the future. However as it's more complicated to reason about, more > > complicated to implement and I'm not entirely sure how much actually needed > > given the (a) `--claim` mode, I think we can wait for feedback from the > > users before actually implementing it. > > > > 6. Enforcing re-upload by a single task and skew > > If we use some greedy logic like subtask 0 always re-uploads then it > > might be overloaded. > > So we'll have to obtain a full list of subtasks first (then probably > > choose randomly or round-robin). > > However, that requires rebuilding Task snapshot, which is doable but > > not trivial (which I think supports "reverse API option"). > > > > What do you mean by "rebuilding Task snapshot"? > > > > During some early discussions about this point, I've hoped that a state > > backend like changelog could embed into the state handle information which > > operator should actually be responsible for duplicating such shared states. > > However now that I'm thinking about it, indeed there might be an issue if > > we combine the fact that state handles can be shared across multiple > > different operators and with a job modification, like dropping an operator. > > In that case it looks like we would need some extra logic during recovery, > > that would have an overview of the whole job to make a decision which > > particular parallel instance of an operator should be responsible for > > duplicating the underlying file? > > > > Best, > > Piotrek > > > > wt., 23 lis 2021 o 12:28 Dawid Wysakowicz <dwysakow...@apache.org> > > <dwysakow...@apache.org> > > napisał(a): > > > > Some user has usage to restore different jobs based on a same externalized > > checkpoint. I think this usage would break after introducing this FLIP, and > > we must tell users explicitly if choose to make Flink manage the > > checkpoints by default. > > > > Could you elaborate what do you mean? The proposal is to use the > > "no-claim" mode by default which should let users start as many jobs as > > they wish from the same externalized checkpoints and it should not cause > > them any harm. Each job effectively will create effectively it's own > > private "copy" of the initial checkpoint along with the 1st taken > > checkpoint. > > > > If the 1st full checkpoint did not complete in the end, the next > > checkpoints have to try to reupload all artifacts again. I think this > > problem could be mitigated if task knows some files have been uploaded > > before. > > > > I don't know how we could achieve that easily. Besides, we have the same > > situation for all, even regular checkpoints don't we? Do we check if e.g. > > diff files has been successfully uploaded in a previous aborted checkpoint? > > I am not saying it's a wrong suggestion, just that I feel it is orthogonal > > and I can't see a straightforward way to implement it. > > > > Best, > > > > Dawid > > On 23/11/2021 07:52, Yun Tang wrote: > > > > Hi, > > > > For the likelihood of never deleting some SST files by RocksDB. > > Unfortunately, it could happen as current level compaction strategy in > > RocksDB is triggered by upper input level size reached to the threshold and > > the compaction priority cannot guarantee all files would be choosed during > > several round compactions. > > > > Actually, I am a bit in favor of this FLIP to manage checkpoints within > > Flink as we have heared from many users that they cannot delete older > > checkpoints after several rounds of re-launching Flink jobs. Current Flink > > would not delete older checkpoints automatically when restoring from older > > retained checkpoint, which makes the base checkpoint directory becomes > > larger and larger. However, if they decide to delete the older checkpoint > > directory of other job-ids, they might not be able to recover from the last > > completed checkpoint as it might depend on some artifacts in older > > checkpoint directory. > > > > And I think reuploading would indeed increase the 1st checkpoint duration > > after restoring. For aliyun oss, the developer said that copping files > > (larger than 32MB) from one location to another within same bucket on DFS > > could cause hundreds millseconds. However, from my experiences, copying on > > HDFS might not be so quick. Maybe some numbers here could be better. > > > > I have two questions here: > > 1. If the 1st full checkpoint did not complete in the end, the next > > checkpoints have to try to reupload all artifacts again. I think this > > problem could be mitigated if task knows some files have been uploaded > > before. > > 2. Some user has usage to restore different jobs based on a same > > externalized checkpoint. I think this usage would break after introducing > > this FLIP, and we must tell users explicitly if choose to make Flink manage > > the checkpoints by default. > > > > Best > > Yun Tang > > > > > > On 2021/11/22 19:49:11 Dawid Wysakowicz wrote: > > > > There is one more fundamental issue with either of your two > > proposals that've just came to my mind. > > What happens if you have externalized checkpoints and the job fails > > before the initial checkpoint can be safely removed? > > > > You start the job from the latest created checkpoint and wait for it > > to be allowed for deletion. Then you can delete it, and all previous > > checkpoints (or am I missing something?) > > > > > > Let me clarify it with an example. You start with chk-42, Flink takes > > e.g. three checkpoints chk-43, chk-44, chk-45 all still reference chk-42 > > files. After that it fails. We have externalized checkpoints enabled, > > therefore we have retained all checkpoints. Users starts a new program > > from let's say chk-45. At this point your proposal does not give the > > user any help in regards when chk-42 can be safely removed. (This is > > also how Flink works right now). > > > > To make it even harder you can arbitrarily complicate it, 1) start a job > > from chk-44, 2) start a job from a chk-47 which depends on chk-45, 3) > > never start a job from chk-44, it is not claimed by any job, thus it is > > never deleted, users must remember themselves that chk-44 originated > > from chk-42 etc.) User would be forced to build a lineage system for > > checkpoints to track which checkpoints depend on each other. > > > > I mean that the design described by FLIP implies the following (PCIIW): > > 1. treat SST files from the initial checkpoint specially: re-upload or > > send placeholder - depending on those attributes in state handle > > 2. (SST files from newer checkpoints are re-uploaded depending on > > confirmation currently; so yes there is tracking, but it's different) > > 3. SharedStateRegistry must allow replacing state under the existing > > key; otherwise, if a new key is used then other parallel subtasks > > should learn somehow this key and use it; However, allowing > > replacement must be limited to this scenario, otherwise it can lead to > > previous checkpoint corruption in normal cases > > > > I might not understand your points, but I don't think FLIP implies any > > of this. The FLIP suggests to send along with the CheckpointBarrier a > > flag "force full checkpoint". Then the state backend should respect it > > and should not use any of the previous shared handles. Now let me > > explain how that would work for RocksDB incremental checkpoints. > > > > 1. Simplest approach: upload all local RocksDB files. This works > > exactly the same as the first incremental checkpoint for a fresh start. > > 2. Improvement on 1) we already do know which files were uploaded for > > the initial checkpoint. Therefore instead of uploading the local > > files that are same with files uploaded for the initial checkpoint > > we call duplicate for those files and upload just the diff. > > > > It does not require any changes to the SharedStateRegistry nor to state > > handles, at least for RocksDB. > > > > Best, > > > > Dawid > > > > > > On 22/11/2021 19:33, Roman Khachatryan wrote: > > > > If you assume the 1st checkpoint needs to be "full" you know you are not > > allowed to use any shared files. > > It's true you should know about the shared files of the previous > > checkpoint, but e.g. RocksDB already tracks that. > > > > I mean that the design described by FLIP implies the following (PCIIW): > > 1. treat SST files from the initial checkpoint specially: re-upload or > > send placeholder - depending on those attributes in state handle > > 2. (SST files from newer checkpoints are re-uploaded depending on > > confirmation currently; so yes there is tracking, but it's different) > > 3. SharedStateRegistry must allow replacing state under the existing > > key; otherwise, if a new key is used then other parallel subtasks > > should learn somehow this key and use it; However, allowing > > replacement must be limited to this scenario, otherwise it can lead to > > previous checkpoint corruption in normal cases > > > > Forcing a full checkpoint after completing N checkpoints instead of > > immediately would only require enabling (1) after N checkpoints. > > And with the "poll API until checkpoint released" approach, those > > changes aren't necessary. > > > > > > There is one more fundamental issue with either of your two proposals > > that've just came to my mind. > > What happens if you have externalized checkpoints and the job fails before > > the initial checkpoint can be safely removed? > > > > You start the job from the latest created checkpoint and wait for it > > to be allowed for deletion. Then you can delete it, and all previous > > checkpoints (or am I missing something?) > > > > > > With tracking the shared files on JM you can not say if you can clear the > > files after couple of checkpoints or 10s, 100s or 1000s, > > which translates into minutes/hours/days/weeks of processing. > > > > This doesn't necessarily translate into higher cost (because of saved > > RPC etc., as I mentioned above). > > However, I do agree that an infinite or arbitrary high delay is > > unacceptable. > > > > The added complexity above doesn't seem negligible to me (especially > > in SharedStateHandle); and should therefore be weighted against those > > operational disadvantages (given that the number of checkpoints to > > wait is bounded in practice). > > > > Regards, > > Roman > > > > > > > > > > On Mon, Nov 22, 2021 at 5:05 PM Dawid Wysakowicz <dwysakow...@apache.org> > > <dwysakow...@apache.org> <dwysakow...@apache.org> <dwysakow...@apache.org> > > wrote: > > > > There is one more fundamental issue with either of your two proposals > > that've just came to my mind. What happens if you have externalized > > checkpoints and the job fails before the initial checkpoint can be > > safely removed? You have a situation where you have a retained > > checkpoint that was built on top of the original one. Basically ending > > in a situation we have right now that you never know when it is safe to > > delete a retained checkpoint. > > > > BTW, the intention for the "claim" mode was to support cases when users > > are concerned with the performance of the first checkpoint. In those > > cases they can claim the checkpoint on don't pay the additional cost of > > the first checkpoint. > > > > Best, > > > > Dawid > > > > On 22/11/2021 14:09, Roman Khachatryan wrote: > > > > Thanks Dawid, > > > > Regarding clarity, > > I think that all proposals require waiting for some event: re-upload / > > checkpoint completion / api response. > > But with the current one, there is an assumption: "initial checkpoint > > can be deleted once a new one completes" (instead of just "initial > > checkpoint can be deleted once the API says it can be deleted"). > > So I think it's actually more clear to offer this explicit API and rely on > > it. > > > > Regarding delaying the deletion, > > I agree that it can delay deletion, but how important is it? > > Checkpoints are usually stored on relatively cheap storage like S3, so > > some delay shouldn't be an issue (especially taking rounding into > > account); it can even be cheaper or comparable to paying for > > re-upload/duplicate calls. > > > > Infinite delay can be an issue though, I agree. > > Maybe @Yun can clarify the likelihood of never deleting some SST files > > by RocksDB? > > For the changelog backend, old files won't be used once > > materialization succeeds. > > > > Yes, my concern is checkpointing time, but also added complexity: > > > > It would be a bit invasive though, as we would have to somehow keep track > > which files should not be reused on TMs. > > > > I think we need this anyway if we choose to re-upload files once the > > job is running. > > The new checkpoint must be formed by re-uploaded old artifacts AND > > uploaded new artifacts. > > > > > > Regards, > > Roman > > > > > > On Mon, Nov 22, 2021 at 12:42 PM Dawid Wysakowicz<dwysakow...@apache.org> > > <dwysakow...@apache.org> <dwysakow...@apache.org> <dwysakow...@apache.org> > > wrote: > > > > @Yun > > > > I think it is a good comment with I agree in principal. However, we use > > --fromSavepoint (cli), savepointPath (REST API), SavepointRestoreSettings > > for both restoring from a savepoint and an externalized checkpoint already. > > I wanted to voice that concern. Nevertheless I am fine with changing it to > > execution.restore-mode, if there are no other comments on that matter, I > > will change it. > > > > @Roman: > > > > Re 1. Correct, stop-with-savepoint should commit side-effects. Will add > > that to the doc. > > > > Re.2 What I don't like about this counter proposal is that it still has no > > clearly defined point in time when it is safe to delete the original > > checkpoint. Users would have a hard time reasoning about it and debugging. > > Even worse, I think worst case it might never happen that all the original > > files are no longer in use (I am not too familiar with RocksDB compaction, > > but what happens if there are key ranges that are never accessed again?) I > > agree it is unlikely, but possible, isn't it? Definitely it can take a > > significant time and many checkpoints to do so. > > > > Re. 3 I believe where you are coming from is that you'd like to keep the > > checkpointing time minimal and reuploading files may increase it. The > > proposal so far builds on the assumption we could in most cases use a cheap > > duplicate API instead of re-upload. I could see this as a follow-up if it > > becomes a bottleneck. It would be a bit invasive though, as we would have > > to somehow keep track which files should not be reused on TMs. > > > > Re. 2 & 3 Neither of the counter proposals work well for taking incremental > > savepoints. We were thinking of building incremental savepoints on the same > > concept. I think delaying the completion of an independent savepoint to a > > closer undefined future is not a nice property of savepoints. > > > > Re 4. Good point. We should make sure the first completed checkpoint has > > the independent/full checkpoint property rather than just the first > > triggered. > > > > Re. 5 & 6 I need a bit more time to look into it. > > > > Best, > > > > Dawid > > > > On 22/11/2021 11:40, Roman Khachatryan wrote: > > > > Hi, > > > > Thanks for the proposal Dawid, I have some questions and remarks: > > > > 1. How will stop-with-savepoint be handled? > > Shouldn't side effects be enforced in this case? (i.e. send > > notifyCheckpointComplete) > > > > 2. Instead of forcing re-upload, can we "inverse control" in no-claim mode? > > Anyways, any external tool will have to poll Flink API waiting for the > > next (full) checkpoint, before deleting the retained checkpoint, > > right? > > Instead, we can provide an API which tells whether the 1st checkpoint > > is still in use (and not force re-upload it). > > > > Under the hood, it can work like this: > > - for the checkpoint Flink recovers from, remember all shared state > > handles it is adding > > - when unregistering shared state handles, remove them from the set above > > - when the set becomes empty the 1st checkpoint can be deleted externally > > > > Besides not requiring re-upload, it seems much simpler and less invasive. > > On the downside, state deletion can be delayed; but I think this is a > > reasonable trade-off. > > > > 3. Alternatively, re-upload not necessarily on 1st checkpoint, but > > after a configured number of checkpoints? > > There is a high chance that after some more checkpoints, initial state > > will not be used (because of compaction), > > so backends won't have to re-upload anything (or small part). > > > > 4. Re-uploaded artifacts must not be deleted on checkpoin abortion > > This should be addressed in > > https://issues.apache.org/jira/browse/FLINK-24611. > > If not, I think the FLIP should consider this case. > > > > 5. Enforcing re-upload by a single task and Changelog state backend > > With Changelog state backend, a file can be shared by multiple operators. > > Therefore, getIntersection() is irrelevant here, because operators > > might not be sharing any key groups. > > (so we'll have to analyze "raw" file usage I think). > > > > 6. Enforcing re-upload by a single task and skew > > If we use some greedy logic like subtask 0 always re-uploads then it > > might be overloaded. > > So we'll have to obtain a full list of subtasks first (then probably > > choose randomly or round-robin). > > However, that requires rebuilding Task snapshot, which is doable but > > not trivial (which I think supports "reverse API option"). > > > > 7. I think it would be helpful to list file systems / object stores > > that support "fast" copy (ideally with latency numbers). > > > > Regards, > > Roman > > > > On Mon, Nov 22, 2021 at 9:24 AM Yun Gao <yungao...@aliyun.com.invalid> > > <yungao...@aliyun.com.invalid> <yungao...@aliyun.com.invalid> > > <yungao...@aliyun.com.invalid> wrote: > > > > Hi, > > > > Very thanks Dawid for proposing the FLIP to clarify the ownership for the > > states. +1 for the overall changes since it makes the behavior clear and > > provide users a determined method to finally cleanup savepoints / retained > > checkpoints. > > > > Regarding the changes to the public interface, it seems currently the > > changes are all bound > > to the savepoint, but from the FLIP it seems perhaps we might also need to > > support the claim declaration > > for retained checkpoints like in the cli side[1] ? If so, then might it be > > better to change the option name > > from `execution.savepoint.restore-mode` to something like > > `execution.restore-mode`? > > > > Best, > > Yun > > > > > > [1] > > https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/checkpoints/#resuming-from-a-retained-checkpoint > > > > > > ------------------------------------------------------------------ > > From:Konstantin Knauf <kna...@apache.org> <kna...@apache.org> > > <kna...@apache.org> <kna...@apache.org> > > Send Time:2021 Nov. 19 (Fri.) 16:00 > > To:dev <dev@flink.apache.org> <dev@flink.apache.org> <dev@flink.apache.org> > > <dev@flink.apache.org> > > Subject:Re: [DISCUSS] FLIP-193: Snapshots ownership > > > > Hi Dawid, > > > > Thanks for working on this FLIP. Clarifying the differences and > > guarantees around savepoints and checkpoints will make it easier and safer > > for users and downstream projects and platforms to work with them. > > > > +1 to the changing the current (undefined) behavior when recovering from > > retained checkpoints. Users can now choose between claiming and not > > claiming, which I think will make the current mixed behavior obsolete. > > > > Cheers, > > > > Konstantin > > > > On Fri, Nov 19, 2021 at 8:19 AM Dawid Wysakowicz <dwysakow...@apache.org> > > <dwysakow...@apache.org> <dwysakow...@apache.org> <dwysakow...@apache.org> > > wrote: > > > > Hi devs, > > > > I'd like to bring up for a discussion a proposal to clean up ownership > > of snapshots, both checkpoints and savepoints. > > > > The goal here is to make it clear who is responsible for deleting > > checkpoints/savepoints files and when can that be done in a safe manner. > > > > Looking forward for your feedback! > > > > Best, > > > > Dawid > > > > [1] https://cwiki.apache.org/confluence/x/bIyqCw > > > > > > > > -- > > > > Konstantin Knaufhttps://twitter.com/snntrablehttps://github.com/knaufk > > > > > > -- > > Konstantin Knauf > > https://twitter.com/snntrable > > https://github.com/knaufk
