Thanks for writing this FLIP Dawid. Just to clarify one thing for the support of forced full snapshots. If a state backend does not support this feature, then the user either has to copy the snapshot manually or resume using --claim mode, create a savepoint in canonical format and then change the state backend if he wants to use --no-claim, right?
Cheers, Till On Fri, Nov 26, 2021 at 11:49 AM Dawid Wysakowicz <dwysakow...@apache.org> wrote: > - maybe include "checkpoint" in mode names, i.e. --no-claim-checkpoint? > > I don't think this is a good idea. The modes apply to both savepoints and > checkpoints, plus it's much longer to type in. (minor) > > - add an explicit option to preserve the current behavior (no claim > and no duplicate)? > > We had an offline discussion about it and so far we were leaning towards > keeping the set of supported options minimal. However, if we really think > the old behaviour is useful we can add a --legacy restore mode. cc > @Konstantin @Piotr > > There seems to be a consensus in the discussion, however, I couldn't > find stop-with-savepoint in the document. > > Sorry, I forgot, about this one. I added a note that savepoints generated > from stop-with-savepoint should commit side effects. > > And I still think it would be nice to list object stores which support > duplicate operation. > > I listed a couple of file systems that do have some sort of a COPY API. > > Best, > > Dawid > On 26/11/2021 11:03, Roman Khachatryan wrote: > > 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> > <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> > <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> <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> <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> <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> > <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> > <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> <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> <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> <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> > <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 > >
