[
https://issues.apache.org/jira/browse/FLINK-10930?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16698560#comment-16698560
]
Yun Tang commented on FLINK-10930:
----------------------------------
[~StephanEwen] I know this suggested changes would impact your original
motivation. However, once you would delete a directory in DFS, which could
still be used to store data for other tasks in distributed environment, you
cannot guarantee this exception not happening. I think it's quite easy to
reproduce this phenomenon by just setting checkpoint timeout very short.
Below is the exception I met, if we take a look at the hdfs's audit log, this
exception happens when {{task create state --> JM delete chk directory --> task
fails to write data to stream}}.
{code:java}
AsynchronousException{java.lang.Exception: Could not materialize checkpoint 14
for operator Flat Map -> Flat Map (1/3).}
at
org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointExceptionHandler.tryHandleCheckpointException(StreamTask.java:1186)
at
org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.handleExecutionException(StreamTask.java:978)
at
org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.run(StreamTask.java:915)
at
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1147)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:622)
at java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.Exception: Could not materialize checkpoint 14 for
operator Flat Map -> Flat Map (1/3).
at
org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.handleExecutionException(StreamTask.java:973)
... 6 more
Caused by: java.util.concurrent.ExecutionException: java.io.IOException: Could
not flush and close the file system output stream to
hdfs:/xxx/a105a6971970401f25714efa67000518/chk-14/1fdcf28d-aaba-4591-85c4-cdab804207ba
in order to obtain the stream state handle
at java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.util.concurrent.FutureTask.get(FutureTask.java:192)
at
org.apache.flink.util.FutureUtil.runIfNotDoneAndGet(FutureUtil.java:53)
at
org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer.<init>(OperatorSnapshotFinalizer.java:61)
at
org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.run(StreamTask.java:884)
... 5 more
Caused by: java.io.IOException: Could not flush and close the file system
output stream to
hdfs:/xxx/a105a6971970401f25714efa67000518/chk-14/1fdcf28d-aaba-4591-85c4-cdab804207ba
in order to obtain the stream state handle
at
org.apache.flink.runtime.state.filesystem.FsCheckpointStreamFactory$FsCheckpointStateOutputStream.closeAndGetHandle(FsCheckpointStreamFactory.java:341)
at
org.apache.flink.runtime.state.DuplicatingCheckpointOutputStream.closeAndGetPrimaryHandle(DuplicatingCheckpointOutputStream.java:268)
at
org.apache.flink.runtime.state.CheckpointStreamWithResultProvider$PrimaryAndSecondaryStream.closeAndFinalizeCheckpointStreamResult(CheckpointStreamWithResultProvider.java:115)
at
org.apache.flink.contrib.streaming.state.RocksDBIncrementalSnapshotOperation.materializeMetaData(RocksDBIncrementalSnapshotOperation.java:383)
at
org.apache.flink.contrib.streaming.state.RocksDBIncrementalSnapshotOperation.runSnapshot(RocksDBIncrementalSnapshotOperation.java:146)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at
org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer.<init>(OperatorSnapshotFinalizer.java:59)
... 6 more
Caused by: java.io.FileNotFoundException: File does not exist:
/xxx/a105a6971970401f25714efa67000518/chk-14/1fdcf28d-aaba-4591-85c4-cdab804207ba
(inode 60074501) [Lease. Holder: DFSClient_NONMAPREDUCE_-103338934_91,
pending creates: 1]
at
org.apache.hadoop.hdfs.server.namenode.FSNamesystem.checkLease(FSNamesystem.java:2623)
at
org.apache.hadoop.hdfs.server.namenode.FSDirWriteFileOp.analyzeFileState(FSDirWriteFileOp.java:521)
at
org.apache.hadoop.hdfs.server.namenode.FSDirWriteFileOp.validateAddBlock(FSDirWriteFileOp.java:161)
at
org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalBlock(FSNamesystem.java:2503)
at
org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.addBlock(NameNodeRpcServer.java:828)
at
org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.addBlock(ClientNamenodeProtocolServerSideTranslatorPB.java:512)
at
org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java)
at
org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:447)
at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:989)
at org.apache.hadoop.ipc.Server$RpcCall.run(Server.java:846)
at org.apache.hadoop.ipc.Server$RpcCall.run(Server.java:789)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1804)
at org.apache.hadoop.ipc.Server$Handler.run(Server.java:2457)
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at
sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at
sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
at
org.apache.hadoop.ipc.RemoteException.instantiateException(RemoteException.java:121)
at
org.apache.hadoop.ipc.RemoteException.unwrapRemoteException(RemoteException.java:88)
at
org.apache.hadoop.hdfs.DataStreamer.locateFollowingBlock(DataStreamer.java:1738)
at
org.apache.hadoop.hdfs.DataStreamer.nextBlockOutputStream(DataStreamer.java:1536)
at org.apache.hadoop.hdfs.DataStreamer.run(DataStreamer.java:658)
Caused by:
org.apache.hadoop.ipc.RemoteException(java.io.FileNotFoundException): File does
not exist:
/xxx/a105a6971970401f25714efa67000518/chk-14/1fdcf28d-aaba-4591-85c4-cdab804207ba
(inode 60074501) [Lease. Holder: DFSClient_NONMAPREDUCE_-103338934_91,
pending creates: 1]
at
org.apache.hadoop.hdfs.server.namenode.FSNamesystem.checkLease(FSNamesystem.java:2623)
at
org.apache.hadoop.hdfs.server.namenode.FSDirWriteFileOp.analyzeFileState(FSDirWriteFileOp.java:521)
at
org.apache.hadoop.hdfs.server.namenode.FSDirWriteFileOp.validateAddBlock(FSDirWriteFileOp.java:161)
at
org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalBlock(FSNamesystem.java:2503)
at
org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.addBlock(NameNodeRpcServer.java:828)
at
org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.addBlock(ClientNamenodeProtocolServerSideTranslatorPB.java:512)
at
org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java)
at
org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:447)
at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:989)
at org.apache.hadoop.ipc.Server$RpcCall.run(Server.java:846)
at org.apache.hadoop.ipc.Server$RpcCall.run(Server.java:789)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1804)
at org.apache.hadoop.ipc.Server$Handler.run(Server.java:2457)
at org.apache.hadoop.ipc.Client.getRpcResponse(Client.java:1481)
at org.apache.hadoop.ipc.Client.call(Client.java:1427)
at org.apache.hadoop.ipc.Client.call(Client.java:1337)
at
org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:227)
at
org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:116)
at com.sun.proxy.$Proxy10.addBlock(Unknown Source)
at
org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.addBlock(ClientNamenodeProtocolTranslatorPB.java:440)
at sun.reflect.GeneratedMethodAccessor24.invoke(Unknown Source)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at
org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:398)
at
org.apache.hadoop.io.retry.RetryInvocationHandler$Call.invokeMethod(RetryInvocationHandler.java:163)
at
org.apache.hadoop.io.retry.RetryInvocationHandler$Call.invoke(RetryInvocationHandler.java:155)
at
org.apache.hadoop.io.retry.RetryInvocationHandler$Call.invokeOnce(RetryInvocationHandler.java:95)
at
org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:335)
at com.sun.proxy.$Proxy11.addBlock(Unknown Source)
at
org.apache.hadoop.hdfs.DataStreamer.locateFollowingBlock(DataStreamer.java:1733)
... 2 more
{code}
> Refactor checkpoint directory layout
> ------------------------------------
>
> Key: FLINK-10930
> URL: https://issues.apache.org/jira/browse/FLINK-10930
> Project: Flink
> Issue Type: Improvement
> Components: State Backends, Checkpointing
> Affects Versions: 1.8.0
> Reporter: Yun Tang
> Assignee: Yun Tang
> Priority: Major
> Fix For: 1.8.0
>
>
> The current checkpoint directory layout is introduced from FLINK-8531 with
> three different scopes for tasks:
> * *EXCLUSIVE* is for state that belongs to one checkpoint only, meta data
> and operator state files.
> * *SHARED* is for state that is possibly part of multiple checkpoints
> * *TASKOWNED* is for state that must never by dropped by the jobManager.
> {code:java}
> /user-defined-dir/{job-id}
> |
> +-- shared/
> +-- taskowned/
> +-- chk-1/ // metadata and operator-state files
> +-- chk-2/
> ...{code}
> If we just retain one complete checkpoint, the expected exclusive directory,
> which is the {{chk-id}} checkpoint directory, should only be one left.
> However, as FLINK-10855 interpreted, the failed/expired checkpoint
> directories would also be left. This is really confusing for users who [uses
> externalized checkpoint to resume
> job|https://ci.apache.org/projects/flink/flink-docs-release-1.6/ops/state/checkpoints.html#resuming-from-a-retained-checkpoint],
> not to mention the checkpoint directory resource leak.
> As far as I could know, if the {{chk-id}} checkpoint directory still
> contains the operator state files, I have no idea how to clean the useless
> {{chk-id}} checkpoint directory gracefully. Once job manager dispose the
> failed/expired checkpoint, the target {{chk-id}} checkpoint directory would
> be deleted by JM. However, this directory would also be create by tasks who
> having not reported to JM. When {{checkpoint coordinator}} received those
> late expired tasks, it would discard those useless handles. However, if JM
> also plans to delete the empty parent folder, which is already unsupported
> after FLINK-8540, another task uploading operator state files would meet
> exception due to its writing target's parent directory has just been removed.
> Currently, we handle task checkpoint failure as task failure and the whole
> job would failover which is not we want.
> From what I see, I plan to separate *EXCLUSIVE* directory into two kind of
> exclusive directories, one is still several {{chk-id}} checkpoint directories
> but only contains its exclusive {{meta data}}, the other is just one
> directory named {{exclusive}} which containing the operator state files.
> Operator state files are exclusive to just one specified checkpoint, we could
> also add {{checkpoint-id}} within their file name to let users easily clean
> up.
> The refactored directory layout should be :
> {code:java}
> /user-defined-dir/{job-id}
> |
> +-- shared/
> +-- taskowned/
> +-- exclusive/ // operator state files
> +-- chk-1/ // metadata
> +-- chk-2/
> ...{code}
>
> This new directory layout would not affect users who use external checkpoint
> to resume jobs, since they still just give
> {{/user-defined-dir/job-id/chk-id}} path to resume job.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
