apache/flink · error · IllegalStateException

Truncation handle has not been initialized

Error message

Truncation handle has not been initialized

What it means

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.

Source

Thrown at flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java:202

            throws IOException {
        if (!HadoopUtils.isMinHadoopVersion(2, 7)) {
            throw new IllegalStateException(
                    "Truncation is not available in hadoop version < 2.7 , You are on Hadoop "
                            + VersionInfo.getVersion());
        }

        if (truncateHandle != null) {
            try {
                return (Boolean) truncateHandle.invoke(hadoopFs, file, length);
            } catch (InvocationTargetException e) {
                ExceptionUtils.rethrowIOException(e.getTargetException());
            } catch (Throwable t) {
                throw new IOException(
                        "Truncation of file failed because of access/linking problems with Hadoop's truncate call. "
                                + "This is most likely a dependency conflict or class loading problem.");
            }
        } else {
            throw new IllegalStateException("Truncation handle has not been initialized");
        }
        return false;
    }

    // ------------------------------------------------------------------------
    //  Committer
    // ------------------------------------------------------------------------

    /**
     * Implementation of a committer for the Hadoop File System abstraction. This implementation
     * commits by renaming the temp file to the final file path. The temp file is truncated before
     * renaming in case there is trailing garbage data.
     */
    static class HadoopFsCommitter implements Committer {

        private final FileSystem fs;
        private final HadoopFsRecoverable recoverable;

View on GitHub (pinned to 2f3c205e92)

Solutions

  1. Upgrade the Hadoop cluster/classpath to 2.7+ so FileSystem.truncate exists and the handle initializes
  2. Set the sink's RollingPolicy to roll on every checkpoint (rolloverOnCheckpoint), avoiding the persist->truncate path during recovery
  3. Verify only one consistent Hadoop version is on the classpath (no shaded/duplicate hadoop-common jars)
  4. (Code-level) Check the truncateHandle/constructor state before recovery and fail with a clearer message

Example fix

// before
// rolling policy that keeps parts open across checkpoints -> truncation needed on recovery
FileSink.forRowFormat(path, encoder)
    .withRollingPolicy(DefaultRollingPolicy.builder().build());

// after
// roll on every checkpoint: no in-progress part survives, no truncation during commit
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicy.CheckpointRollingPolicy;
FileSink.forRowFormat(path, encoder)
    .withRollingPolicy(new CheckpointRollingPolicy<String, String>() {
        public boolean shouldRollOnEvent(PartFileInfo info, String line) { return false; }
        public boolean shouldRollOnProcessingTime(PartFileInfo info, long t) { return false; }
    });
Defensive patterns

Strategy: validation

Validate before calling

// Before recovery, assert truncate is available on this Hadoop
org.apache.hadoop.fs.FileSystem hfs = ...;
boolean hasTruncate;
try {
    hfs.getClass().getMethod("truncate", org.apache.hadoop.fs.Path.class, long.class);
    hasTruncate = true;
} catch (NoSuchMethodException e) {
    hasTruncate = false;
}
if (!hasTruncate) {
    // use a RollingPolicy that rolls on every checkpoint so commitAfterRecovery never truncates
}

Try / catch

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; }

Prevention

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/fb638d0a8a4fc3da. Report an issue: GitHub.