{"record":{"id":"998a9677b0e8a95e","repo":"apache/flink","slug":"in-progress-file-s-not-exists","errorCode":null,"errorMessage":"In progress file(%s) not exists.","messagePattern":"In progress file\\((.+?)\\) not exists\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-hadoop-bulk/src/main/java/org/apache/flink/formats/hadoop/bulk/committer/HadoopRenameFileCommitter.java","lineNumber":90,"sourceCode":"        // Do nothing.\n    }\n\n    @Override\n    public void commit() throws IOException {\n        rename(true);\n    }\n\n    @Override\n    public void commitAfterRecovery() throws IOException {\n        rename(false);\n    }\n\n    private void rename(boolean assertFileExists) throws IOException {\n        FileSystem fileSystem = FileSystem.get(targetFilePath.toUri(), configuration);\n\n        if (!fileSystem.exists(tempFilePath)) {\n            if (assertFileExists) {\n                throw new IOException(\n                        String.format(\"In progress file(%s) not exists.\", tempFilePath));\n            } else {\n                // By pass the re-commit if source file not exists.\n                // TODO: in the future we may also need to check if the target file exists.\n                return;\n            }\n        }\n\n        try {\n            // If file exists, it will be overwritten.\n            fileSystem.rename(tempFilePath, targetFilePath);\n        } catch (IOException e) {\n            throw new IOException(\n                    String.format(\n                            \"Could not commit file from %s to %s\", tempFilePath, targetFilePath),\n                    e);\n        }\n    }","sourceCodeStart":72,"sourceCodeEnd":108,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-hadoop-bulk/src/main/java/org/apache/flink/formats/hadoop/bulk/committer/HadoopRenameFileCommitter.java#L72-L108","documentation":"Thrown by HadoopRenameFileCommitter.commit() when the pending (temp) file no longer exists on the target FileSystem at commit time. The committer writes to a hidden temp path next to the target and renames it on commit; if the temp file vanished before the rename, the pre-commit existence assertion fails. This typically indicates the commit is being retried after the file was already moved, deleted by an external process, or lost due to eventual-consistency effects on object stores.","triggerScenarios":"Calling StreamingFileSink/FileSink commit (rename(true)) when tempFilePath does not exist: FileSink pending-file cleanup ran early, another committer instance already renamed the file, an external job/ cleanser deleted in-progress dot-files, or S3-style stores where exists() is stale.","commonSituations":"Recovery after a task failure where commit() (not commitAfterRecovery()) is replayed; two sink subtasks writing the same target path; lifecycle policies on HDFS/S3 deleting hidden in-progress files; misconfigured part-file prefix/suffix colliding with cleanup rules.","solutions":["Verify only one sink/subtask owns the target path and the part-file path is unique (partIndex / bucket assigner)","Check no external cleaner deletes hidden in-progress files (temp name starts with '.' next to the target)","On failure-recovery paths make sure the framework calls commitAfterRecovery() (which bypasses a missing source file) rather than a fresh commit()","Inspect FileSystem logs around the failure to see who renamed or deleted the temp file","For eventually-consistent object stores, verify exists() is not reading a stale listing before concluding the file is lost"],"exampleFix":"// before: assuming the pending file always survives until commit\ncommitter.commit();\n\n// after: tolerate already-committed state on retry\nif (fs.exists(committer.getTempFilePath())) {\n    committer.commit();\n} else {\n    // file was already moved or deleted; skip or use commitAfterRecovery()\n    committer.commitAfterRecovery();\n}","handlingStrategy":"try-catch","validationCode":"FileSystem fs = FileSystem.get(targetPath.toUri(), conf);\nif (!fs.exists(tempPath)) {\n    // already committed or cleaned; skip fresh commit, use recovery semantics\n}","typeGuard":null,"tryCatchPattern":"try {\n    committer.commit();\n} catch (IOException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"not exists\")) {\n        // treat as already-committed; log and continue idempotently\n    } else {\n        throw e;\n    }\n}","preventionTips":["Guarantee a single writer per target part-file path (unique bucket/part file naming)","Ensure recovery paths invoke commitAfterRecovery() rather than a fresh commit()","Disable lifecycle/cleanup rules that delete hidden in-progress files in sink directories"],"tags":["flink","hadoop","filesystem","file-sink","commit","exactly-once"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}