{"record":{"id":"62b0426fb19b5e0f","repo":"apache/flink","slug":"cannot-clean-commit-file-has-trailing-junk-data-62b042","errorCode":null,"errorMessage":"Cannot clean commit: File has trailing junk data.","messagePattern":"Cannot clean commit: File has trailing junk data\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java","lineNumber":242,"sourceCode":"        }\n\n        @Override\n        public void commit() throws IOException {\n            final Path src = recoverable.tempFile();\n            final Path dest = recoverable.targetFile();\n            final long expectedLength = recoverable.offset();\n\n            final FileStatus srcStatus;\n            try {\n                srcStatus = fs.getFileStatus(src);\n            } catch (IOException e) {\n                throw new IOException(\"Cannot clean commit: Staging file does not exist.\");\n            }\n\n            if (srcStatus.getLen() != expectedLength) {\n                // something was done to this file since the committer was created.\n                // this is not the \"clean\" case\n                throw new IOException(\"Cannot clean commit: File has trailing junk data.\");\n            }\n\n            try {\n                fs.rename(src, dest);\n            } catch (IOException e) {\n                throw new IOException(\n                        \"Committing file by rename failed: \" + src + \" to \" + dest, e);\n            }\n        }\n\n        @Override\n        public void commitAfterRecovery() throws IOException {\n            final Path src = recoverable.tempFile();\n            final Path dest = recoverable.targetFile();\n            final long expectedLength = recoverable.offset();\n\n            FileStatus srcStatus = null;\n            try {","sourceCodeStart":224,"sourceCodeEnd":260,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java#L224-L260","documentation":"During a clean commit, HadoopFsCommitter compares the staging file's current length against recoverable.offset() (the byte position recorded when persist() snapshot the state). A mismatch means bytes were appended to (or truncated from) the temp file after the recoverable was created, so the 'clean' rename-only commit is unsafe and is refused.","triggerScenarios":"persist() is called for a checkpoint, writing continues into the same temp file (buffered data flushed later), and then commit() is invoked with the older recoverable; a commit retried concurrently with an active writer; duplicate consumption of the same checkpointed committer state.","commonSituations":"StreamingFileSink recovery where the operator resumes writing and then commits an older pending checkpoint; at-least-once replay causing the same part file to be appended by two attempts; taking a savepoint while a checkpoint commit is still pending.","solutions":["Ensure only one job instance processes a given in-progress file at a time (proper job restart, not parallel duplicates)","Resume from the latest checkpoint/savepoint so the recoverable offset matches the file tail, or let commitAfterRecovery handle the truncation path instead of clean commit","If unavoidable, discard the mismatched in-progress file and restart the part"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Pre-check the clean-commit invariant: file length must equal recoverable offset\nFileStatus st = fs.getFileStatus(recoverable.tempFile());\nif (st.getLen() != recoverable.offset()) {\n    // route to the recovery-style commit (truncate then rename) instead of clean commit\n}","typeGuard":null,"tryCatchPattern":"catch (IOException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"trailing junk data\")) {\n        // fall back to truncate-then-rename semantics or discard the part\n    } else { throw e; }\n}","preventionTips":["One active writer per output path/UID at a time","Prefer rolling policies that roll on checkpoint (offset always matches tail)","Avoid replaying old committer state alongside a running job"],"tags":["hdfs","file-sink","commit","consistency","recovery"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}