{"record":{"id":"fb638d0a8a4fc3da","repo":"apache/flink","slug":"truncation-handle-has-not-been-initialized","errorCode":null,"errorMessage":"Truncation handle has not been initialized","messagePattern":"Truncation handle has not been initialized","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java","lineNumber":202,"sourceCode":"            throws IOException {\n        if (!HadoopUtils.isMinHadoopVersion(2, 7)) {\n            throw new IllegalStateException(\n                    \"Truncation is not available in hadoop version < 2.7 , You are on Hadoop \"\n                            + VersionInfo.getVersion());\n        }\n\n        if (truncateHandle != null) {\n            try {\n                return (Boolean) truncateHandle.invoke(hadoopFs, file, length);\n            } catch (InvocationTargetException e) {\n                ExceptionUtils.rethrowIOException(e.getTargetException());\n            } catch (Throwable t) {\n                throw new IOException(\n                        \"Truncation of file failed because of access/linking problems with Hadoop's truncate call. \"\n                                + \"This is most likely a dependency conflict or class loading problem.\");\n            }\n        } else {\n            throw new IllegalStateException(\"Truncation handle has not been initialized\");\n        }\n        return false;\n    }\n\n    // ------------------------------------------------------------------------\n    //  Committer\n    // ------------------------------------------------------------------------\n\n    /**\n     * Implementation of a committer for the Hadoop File System abstraction. This implementation\n     * commits by renaming the temp file to the final file path. The temp file is truncated before\n     * renaming in case there is trailing garbage data.\n     */\n    static class HadoopFsCommitter implements Committer {\n\n        private final FileSystem fs;\n        private final HadoopFsRecoverable recoverable;\n","sourceCodeStart":184,"sourceCodeEnd":220,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java#L184-L220","documentation":"HadoopRecoverableFsDataOutputStream reflects Hadoop FileSystem's truncate(Path, long) method into a Method handle at construction time. If the reflective lookup fails (Hadoop version without truncate, primarily < 2.7, or classloading issues), truncateHandle stays null. Calling truncate (directly or via commitAfterRecovery on a file with trailing data) then throws this IllegalStateException.","triggerScenarios":"commitAfterRecovery() on a staging file longer than recoverable.offset() (persist -> recovery transition), on a Hadoop distribution whose FileSystem class lacks a truncate(Path,long) method, or when the Hadoop classes seen by Flink's classloader differ from the runtime one so getDeclaredMethod fails.","commonSituations":"Running the StreamingFileSink/FileSink with HDFS on Hadoop 2.6 or an old vendor-shaded Hadoop; mixing Hadoop versions between Flink's bundled hadoop-fs and the cluster's Hadoop; a rolling policy that does not roll on checkpoint causing truncation to be exercised during recovery.","solutions":["Upgrade the Hadoop cluster/classpath to 2.7+ so FileSystem.truncate exists and the handle initializes","Set the sink's RollingPolicy to roll on every checkpoint (rolloverOnCheckpoint), avoiding the persist->truncate path during recovery","Verify only one consistent Hadoop version is on the classpath (no shaded/duplicate hadoop-common jars)","(Code-level) Check the truncateHandle/constructor state before recovery and fail with a clearer message"],"exampleFix":"// before\n// rolling policy that keeps parts open across checkpoints -> truncation needed on recovery\nFileSink.forRowFormat(path, encoder)\n    .withRollingPolicy(DefaultRollingPolicy.builder().build());\n\n// after\n// roll on every checkpoint: no in-progress part survives, no truncation during commit\nimport org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicy.CheckpointRollingPolicy;\nFileSink.forRowFormat(path, encoder)\n    .withRollingPolicy(new CheckpointRollingPolicy<String, String>() {\n        public boolean shouldRollOnEvent(PartFileInfo info, String line) { return false; }\n        public boolean shouldRollOnProcessingTime(PartFileInfo info, long t) { return false; }\n    });","handlingStrategy":"validation","validationCode":"// Before recovery, assert truncate is available on this Hadoop\norg.apache.hadoop.fs.FileSystem hfs = ...;\nboolean hasTruncate;\ntry {\n    hfs.getClass().getMethod(\"truncate\", org.apache.hadoop.fs.Path.class, long.class);\n    hasTruncate = true;\n} catch (NoSuchMethodException e) {\n    hasTruncate = false;\n}\nif (!hasTruncate) {\n    // use a RollingPolicy that rolls on every checkpoint so commitAfterRecovery never truncates\n}","typeGuard":null,"tryCatchPattern":"catch (IllegalStateException e) { /* message: Truncation handle has not been initialized */ LOG.error(\"Hadoop lacks truncate(Path,long); upgrade Hadoop or roll on checkpoint\", e); throw e; }","preventionTips":["Run Hadoop 2.7+ for file sinks with recovery","Configure the sink's rolling policy to roll on every checkpoint","Keep one consistent Hadoop version across the Flink classpath"],"tags":["hadoop","hdfs","file-sink","truncation","reflection","streaming-files"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}