The issue with corrupted files is that some of them aren't read by the
component that stores them.
For example, a file can be corrupted in the blob server of the JM, but
that it is corrupted will only be noticed by the TaskExecutor.
On 15/12/2021 11:36, Till Rohrmann wrote:
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
