Thanks everyone for your feedback. Let me try to address it by grouping some of the individual comments:
### Will this feature work for native Yarn and K8s deployments? The working directory is an optional feature that can be used to recover additional information. You can think of it like a cache. If the working directory is there, then Flink can do certain things a bit faster but in the worst case it will have to retrieve the required information from the JobManager or persistent storage. In order to make it work with native Yarn and K8s, we would have to change these modes slightly. First of all, we would have to be able to map working directories to processes and then set a deterministic resource ids for the processes. For K8s this could be easily achievable by using a StatefulSet as the deployment mechanism for TaskExecutors. For Yarn, we probably would have to remember the prior locations of a process. Both things are potential follow ups that I don't want to tackle in this FLIP. If one of the modes configures the working directory to be on a full or broken disk, then the process will fail. I think this is not all that different from the current state where some things in Flink will fail if they picked the wrong/full temporary directory (e.g. blob storage directory). ### Cleanup The working directory will be cleaned up if the Flink process is gracefully shut down. This means that the JobManager process will clean it up if it runs in application mode and the job is terminated. SIGTERM and SIGKILL signals will be treated as an ungraceful shutdown and therefore they won't clean up the working directory. This means that we probably also need a graceful way for shutting TaskManager processes down in the future because right now they are in most cases killed in order to shut them down. If the user uses the tmp directory, then any left-over working directories will be cleaned up with the next system restart. This is somewhat similar to how RocksDB's working directory is currently cleaned up as well. ### Corrupted files The working directory itself won't give you any guarantees. It will be the responsibility of the component that uses the working directory to make sure that it can deal with corrupted files. E.g. if the component cannot read the file, then it should delete it and fall back to the remote storage/ground truth to retrieve the required information. I hope this could answer your questions. Let me know if you have more feedback. Cheers, Till On Mon, Dec 13, 2021 at 5:05 AM 刘建刚 <liujiangangp...@gmail.com> wrote: > I like the idea. It can reuse the disk to do many things. Isn't it only > for inner failover? If not, the cleaning may be a problem. Also, many > resource components have their own disk schedule strategy. > > Chesnay Schepler <ches...@apache.org> 于2021年12月12日周日 19:59写道： > >> How do you intend to handle corrupted files, in particular due to >> process crashes during a write? >> Will all writes to a cached directory append some suffix (e.g., >> ".pending") and do a rename? >> >> On 10/12/2021 17:54, Till Rohrmann wrote: >> > Hi everyone, >> > >> > I would like to start a discussion about introducing an explicit working >> > directory for Flink processes that can be used to store information [1]. >> > Per default this working directory will reside in the temporary >> directory >> > of the node Flink runs on. However, if configured to reside on a >> persistent >> > volume, then this information can be used to recover from process/node >> > failures. Moreover, such a working directory can be used to consolidate >> > some of our other directories Flink creates under /tmp (e.g. >> blobStorage, >> > RocksDB working directory). >> > >> > Here is a draft PR that outlines the required changes [2]. >> > >> > Looking forward to your feedback. >> > >> > [1] https://cwiki.apache.org/confluence/x/ZZiqCw >> > [2] https://github.com/apache/flink/pull/18083 >> > >> > Cheers, >> > Till >> > >> >>
