[
https://issues.apache.org/jira/browse/FLINK-37686?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18098085#comment-18098085
]
Francis edited comment on FLINK-37686 at 7/22/26 8:23 AM:
----------------------------------------------------------
We hit this in production last week and can add a reproduction pattern and some
analysis.
Environment: Flink 2.2.1 on Kubernetes, ForSt disaggregated state with
state.backend.forst.primary-dir at the default (checkpoint dir), incremental
checkpoints every 10 minutes, S3 via flink-s3-fs-presto, parallelism 10, ~6.4
TB of state, execution.checkpointing.num-retained: 3.
The failure pattern is deterministic and tied to num-retained. After a restore
the job completes exactly 3 checkpoints, then the next one fails with the stack
below, and every subsequent checkpoint fails the same way until the failure
threshold forces a failover, which starts the next cycle. Six consecutive
cycles failed exactly at that boundary. Two later runs got slightly further (6
and 4 completions) before failing the same way, and the job only stabilized
once ingest dropped in the evening. Each cycle cost 30 to 50 minutes of Kafka
reprocessing.
{code}
java.io.FileNotFoundException: File does not exist:
s3p://.../checkpoints/<job>/shared/op_AsyncKeyedProcessOperator_..._attempt_0/db/<uuid>
at
com.facebook.presto.hive.s3.PrestoS3FileSystem.getFileStatus(PrestoS3FileSystem.java:361)
at
org.apache.flink.state.forst.fs.ForStFlinkFileSystem.getFileStatus(ForStFlinkFileSystem.java:340)
at
org.apache.flink.state.forst.fs.ForStFlinkFileSystem.listStatus(ForStFlinkFileSystem.java:376)
at
org.apache.flink.state.forst.fs.StringifiedForStFileSystem.listStatus(StringifiedForStFileSystem.java:52)
at org.forstdb.RocksDB.enableFileDeletions(Native Method)
at
org.apache.flink.state.forst.snapshot.ForStNativeFullSnapshotStrategy.lambda$syncPrepareResources$3(ForStNativeFullSnapshotStrategy.java:188)
{code}
The first failure happens right after the restored checkpoint leaves the
retained set, this looks to be when the JM deletes shared handles whose
reference count reached zero. Some of those objects are still present in the
TM-side FileMappingManager. The next enableFileDeletions full scan stats every
mapping entry via listStatus -> getFileStatus(entry.getSourcePath()) and throws
on the first deleted object. The failure then repeats every checkpoint because
the full scan is the only path that would drop the stale entry, and it dies
before it gets there. A failover clears it because the mapping is rebuilt from
the restored checkpoint, whose files are all still present by the registry's
own accounting.
Two more data points:
- The odds of failures look to increase with write volume. Each restore rewinds
the source and replays at 2 to 3x the normal rate, producing more short-lived
SSTs per, so it looks like the loop feeds itself. Run length correlated
inversely with ingest across the whole incident, and on two prior days we saw
single occurrences at lower load that recovered after one failover.
- On the native side, EnableFileDeletions -> FindObsoleteFiles(force=true)
forces the GetChildren full scan. That call site already tolerates listing
errors (s.PermitUncheckedError()), but the pending Java exception from the JNI
filesystem callback looks to bubble up regardless.
These look to be the relevant classes (ForStFlinkFileSystem,
FileMappingManager, MappingEntry, FileOwnershipDecider,
ReusableDataTransferStrategy) and they seem the same, so no released 2.x avoids
this issue.
Suggested fix: in the listStatus path used by the purge scan, tolerate
FileNotFoundException for entries whose ownership is NOT_OWNED, I think we can
skip them and drop them from the mapping table. Those files are JM-owned and
already deleted, and the purge was about to stop tracking them anyway. I can
put up a PR with a test if that direction sounds right.
Operational workaround for anyone else hitting this: creating a zero-byte
object at the exact key from the exception unblocks the next scan. One stale
entry is purged per checkpoint, so it can take a few placeholders. We ended up
needing to add about 15 to work past this issue. The object is never read, so
an empty file looks to be OK. Moving state.backend.forst.primary-dir off the
checkpoint directory avoids the ownership handover entirely, at the documented
cost of file copies during checkpointing and recovery. Given our state store
size this is unfortunately not an option for us. Hope this helps!
[~masteryhx] Does this make sense to you at all?
was (Author: JIRAUSER305300):
We hit this in production last week and can add a reproduction pattern and some
analysis.
Environment: Flink 2.2.1 on Kubernetes, ForSt disaggregated state with
state.backend.forst.primary-dir at the default (checkpoint dir), incremental
checkpoints every 10 minutes, S3 via flink-s3-fs-presto, parallelism 10, ~6.4
TB of state, execution.checkpointing.num-retained: 3.
The failure pattern is deterministic and tied to num-retained. After a restore
the job completes exactly 3 checkpoints, then the next one fails with the stack
below, and every subsequent checkpoint fails the same way until the failure
threshold forces a failover, which starts the next cycle. Six consecutive
cycles failed exactly at that boundary. Two later runs got slightly further (6
and 4 completions) before failing the same way, and the job only stabilized
once ingest dropped in the evening. Each cycle cost 30 to 50 minutes of Kafka
reprocessing.
{code}
java.io.FileNotFoundException: File does not exist:
s3p://.../checkpoints/<job>/shared/op_AsyncKeyedProcessOperator_..._attempt_0/db/<uuid>
at
com.facebook.presto.hive.s3.PrestoS3FileSystem.getFileStatus(PrestoS3FileSystem.java:361)
at
org.apache.flink.state.forst.fs.ForStFlinkFileSystem.getFileStatus(ForStFlinkFileSystem.java:340)
at
org.apache.flink.state.forst.fs.ForStFlinkFileSystem.listStatus(ForStFlinkFileSystem.java:376)
at
org.apache.flink.state.forst.fs.StringifiedForStFileSystem.listStatus(StringifiedForStFileSystem.java:52)
at org.forstdb.RocksDB.enableFileDeletions(Native Method)
at
org.apache.flink.state.forst.snapshot.ForStNativeFullSnapshotStrategy.lambda$syncPrepareResources$3(ForStNativeFullSnapshotStrategy.java:188)
{code}
The first failure happens right after the restored checkpoint leaves the
retained set, this looks to be when the JM deletes shared handles whose
reference count reached zero. Some of those objects are still present in the
TM-side FileMappingManager. The next enableFileDeletions full scan stats every
mapping entry via listStatus -> getFileStatus(entry.getSourcePath()) and throws
on the first deleted object. The failure then repeats every checkpoint because
the full scan is the only path that would drop the stale entry, and it dies
before it gets there. A failover clears it because the mapping is rebuilt from
the restored checkpoint, whose files are all still present by the registry's
own accounting.
Two more data points:
- The odds of failures look to increase with write volume. Each restore rewinds
the source and replays at 2 to 3x the normal rate, producing more short-lived
SSTs per, so it looks like the loop feeds itself. Run length correlated
inversely with ingest across the whole incident, and on two prior days we saw
single occurrences at lower load that recovered after one failover.
- On the native side, EnableFileDeletions -> FindObsoleteFiles(force=true)
forces the GetChildren full scan. That call site already tolerates listing
errors (s.PermitUncheckedError()), but the pending Java exception from the JNI
filesystem callback looks to bubble up regardless.
These look to be the relevant classes (ForStFlinkFileSystem,
FileMappingManager, MappingEntry, FileOwnershipDecider,
ReusableDataTransferStrategy) and they seem the same, so no released 2.x avoids
this issue.
Suggested fix: in the listStatus path used by the purge scan, tolerate
FileNotFoundException for entries whose ownership is NOT_OWNED, I think we can
skip them and drop them from the mapping table. Those files are JM-owned and
already deleted, and the purge was about to stop tracking them anyway. I can
put up a PR with a test if that direction sounds right.
Operational workaround for anyone else hitting this: creating a zero-byte
object at the exact key from the exception unblocks the next scan. One stale
entry is purged per checkpoint, so it can take a few placeholders. We ended up
needing to add about 15 to work past this issue. The object is never read, so
an empty file looks to be OK. Moving state.backend.forst.primary-dir off the
checkpoint directory avoids the ownership handover entirely, at the documented
cost of file copies during checkpointing and recovery. Given our state store
size this is unfortunately not an option for us. Hope this helps!
> ForSt consecutive failed checkpoint due to check deleted file status when
> enableFileDeletions
> ---------------------------------------------------------------------------------------------
>
> Key: FLINK-37686
> URL: https://issues.apache.org/jira/browse/FLINK-37686
> Project: Flink
> Issue Type: Bug
> Components: Runtime / State Backends
> Affects Versions: 2.1.0
> Reporter: Hangxiang Yu
> Assignee: Hangxiang Yu
> Priority: Major
> Attachments: EnableFileDeletions.png, Full Scan Obsolete Files.png
>
>
> Currently, ForSt will try to reuse files when checkpoint.
> When reuse, FileOwnership is NOT_OWNED
>
> {code:java}
> 2025-04-16 13:03:55,952 INFO
> org.apache.flink.state.forst.fs.filemapping.FileMappingManager [] - Give up
> ownership for file: MappingEntry{source=HandleBackedSource{stateHandle=File
> State:
> hdfs://k8s-flink-test/checkpoint/39779093/shared/op_KeyedProcessOperator_90bea66de1c231edf33913ecd54406c1__1_1__attempt_0/db/cfc1d618-6e02-42f7-8718-553f23f6e08e
> [52273557 bytes]}, fileOwnership=NOT_OWNED, isDirectory= false}, the source
> is now backed by: File State:
> hdfs://k8s-flink-test/checkpoint/39779093/shared/op_KeyedProcessOperator_90bea66de1c231edf33913ecd54406c1__1_1__attempt_0/db/cfc1d618-6e02-42f7-8718-553f23f6e08e
> [52273557 bytes] {code}
>
> and this file maybe deleted by JM in async checkpoint cleaner thread.
>
> {code:java}
> 2025-04-16 13:07:01,920 INFO
> org.apache.flink.runtime.state.SharedStateRegistryImpl [] - Scheduled
> delete of state handle File State:
> hdfs://k8s-flink-test/checkpoint/39779093/shared/op_KeyedProcessOperator_90bea66de1c231edf33913ecd54406c1__1_1__attempt_0/db/cfc1d618-6e02-42f7-8718-553f23f6e08e
> [52273557 bytes].{code}
>
> But it may still be recorded in ForSt ObsoleteFiles, everytime
> enableFileDeletions is called, ForSt will check whether all obsolete files
> are existed, and may fails if it's deleted by JM.
> {code:java}
> 2025-04-16 13:07:53,123 INFO
> org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Triggering
> checkpoint 8 (type=CheckpointType{name='Checkpoint',
> sharingFilesStrategy=FORWARD_BACKWARD}) @ 1744780073120 for job
> e3fca74b64c70018276be5ab859a2ce2.
> 2025-04-16 13:08:04,848 INFO
> org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Decline
> checkpoint 8 by task
> 8bbfb2b8ab9d60131688ca9a1b38fa4c_90bea66de1c231edf33913ecd54406c1_0_0 of job
> e3fca74b64c70018276be5ab859a2ce2 at
> application-flink-k8s-1744779470362-5783573-taskmanager-1-1.
> org.apache.flink.util.SerializedThrowable:
> org.apache.flink.runtime.checkpoint.CheckpointException: Asynchronous task
> checkpoint failed.
> at
> org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable.handleExecutionException(AsyncCheckpointRunnable.java:320)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable.run(AsyncCheckpointRunnable.java:155)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
> [?:?]
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
> [?:?]
> at java.lang.Thread.run(Thread.java:829) [?:?]
> Caused by: org.apache.flink.util.SerializedThrowable: java.lang.Exception:
> Could not materialize checkpoint 8 for operator GroupAggregate[4] -> Calc[5]
> -> Sink: print_sink[6] (1/1)#0.
> at
> org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable.handleExecutionException(AsyncCheckpointRunnable.java:298)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> ... 4 more
> Caused by: org.apache.flink.util.SerializedThrowable:
> java.util.concurrent.ExecutionException: java.io.FileNotFoundException: File
> does not exist:
> hdfs://k8s-flink-test/checkpoint/39779093/shared/op_KeyedProcessOperator_90bea66de1c231edf33913ecd54406c1__1_1__attempt_0/db/cfc1d618-6e02-42f7-8718-553f23f6e08e
> at java.util.concurrent.FutureTask.report(FutureTask.java:122) ~[?:?]
> at java.util.concurrent.FutureTask.get(FutureTask.java:191) ~[?:?]
> at
> org.apache.flink.util.concurrent.FutureUtils.runIfNotDoneAndGet(FutureUtils.java:511)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer.<init>(OperatorSnapshotFinalizer.java:54)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable.finalizeNonFinishedSnapshots(AsyncCheckpointRunnable.java:191)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable.run(AsyncCheckpointRunnable.java:124)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> ... 3 more
> Caused by: org.apache.flink.util.SerializedThrowable:
> java.io.FileNotFoundException: File does not exist:
> hdfs://k8s-flink-test/checkpoint/39779093/shared/op_KeyedProcessOperator_90bea66de1c231edf33913ecd54406c1__1_1__attempt_0/db/cfc1d618-6e02-42f7-8718-553f23f6e08e
> at
> org.apache.hadoop.hdfs.DistributedFileSystem$24.doCall(DistributedFileSystem.java:1417)
> ~[flink-shaded-hadoop-2-uber-2.7.2-10.7.jar:2.7.2-10.7]
> at
> org.apache.hadoop.hdfs.DistributedFileSystem$24.doCall(DistributedFileSystem.java:1409)
> ~[flink-shaded-hadoop-2-uber-2.7.2-10.7.jar:2.7.2-10.7]
> at
> org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)
> ~[flink-shaded-hadoop-2-uber-2.7.2-10.7.jar:2.7.2-10.7]
> at
> org.apache.hadoop.hdfs.DistributedFileSystem.getFileStatus(DistributedFileSystem.java:1425)
> ~[flink-shaded-hadoop-2-uber-2.7.2-10.7.jar:2.7.2-10.7]
> at
> org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.getFileStatus(HadoopFileSystem.java:85)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.core.fs.SafetyNetWrapperFileSystem.getFileStatus(SafetyNetWrapperFileSystem.java:65)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.state.forst.fs.ForStFlinkFileSystem.getFileStatus(ForStFlinkFileSystem.java:320)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.state.forst.fs.ForStFlinkFileSystem.listStatus(ForStFlinkFileSystem.java:356)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.state.forst.fs.StringifiedForStFileSystem.listStatus(StringifiedForStFileSystem.java:52)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at org.forstdb.RocksDB.enableFileDeletions(Native Method)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at org.forstdb.RocksDB.enableFileDeletions(RocksDB.java:4312)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.state.forst.snapshot.ForStNativeFullSnapshotStrategy.lambda$syncPrepareResources$3(ForStNativeFullSnapshotStrategy.java:188)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.state.forst.snapshot.ForStSnapshotStrategyBase$ForStNativeSnapshotResources.release(ForStSnapshotStrategyBase.java:247)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.state.forst.snapshot.ForStIncrementalSnapshotStrategy$ForStIncrementalSnapshotOperation.get(ForStIncrementalSnapshotStrategy.java:287)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.runtime.state.SnapshotStrategyRunner$1.callInternal(SnapshotStrategyRunner.java:91)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.runtime.state.SnapshotStrategyRunner$1.callInternal(SnapshotStrategyRunner.java:88)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.runtime.state.AsyncSnapshotCallable.call(AsyncSnapshotCallable.java:78)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
> at
> org.apache.flink.util.concurrent.FutureUtils.runIfNotDoneAndGet(FutureUtils.java:508)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer.<init>(OperatorSnapshotFinalizer.java:54)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable.finalizeNonFinishedSnapshots(AsyncCheckpointRunnable.java:191)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> at
> org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable.run(AsyncCheckpointRunnable.java:124)
> ~[flink-dist_2.12-2.0.0-SNAPSHOT.jar:2.0.0-SNAPSHOT]
> ... 3 more{code}
> We should do special check for NOT_OWNED file when enableFileDeletions.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
