apache/flink · error · FlinkRuntimeException

Could not create writer state serializer.

Error message

Could not create writer state serializer.

What it means

Thrown by getWriterStateSerializer() in FileSink when bucketsBuilder.getWriterStateSerializer() fails with IOException. Creating the serializer internally calls createBucketWriter() which invokes FileSystem.get(basePath.toUri()).createRecoverableWriter(), which can fail due to filesystem configuration or connectivity issues. The interface method does not declare IOException, so it is wrapped as FlinkRuntimeException.

Source

Thrown at flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java:170

    @Override
    public FileWriter<IN> restoreWriter(
            WriterInitContext context, Collection<FileWriterBucketState> recoveredState)
            throws IOException {
        FileWriter<IN> writer = bucketsBuilder.createWriter(context);
        writer.initializeState(recoveredState);
        return writer;
    }

    @Override
    public SimpleVersionedSerializer<FileWriterBucketState> getWriterStateSerializer() {
        try {
            return bucketsBuilder.getWriterStateSerializer();
        } catch (IOException e) {
            // it's not optimal that we have to do this but creating the serializers for the
            // FileSink requires (among other things) a call to FileSystem.get() which declares
            // IOException.
            throw new FlinkRuntimeException("Could not create writer state serializer.", e);
        }
    }

    @Override
    public Committer<FileSinkCommittable> createCommitter(CommitterInitContext context)
            throws IOException {
        return bucketsBuilder.createCommitter();
    }

    @Override
    public SimpleVersionedSerializer<FileSinkCommittable> getCommittableSerializer() {
        try {
            return bucketsBuilder.getCommittableSerializer();
        } catch (IOException e) {
            // it's not optimal that we have to do this but creating the serializers for the
            // FileSink requires (among other things) a call to FileSystem.get() which declares
            // IOException.
            throw new FlinkRuntimeException("Could not create committable serializer.", e);

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Verify the basePath URI is correct and the target filesystem is accessible.
2. Ensure the required filesystem plugin is on the classpath (e.g., flink-s3-fs-hadoop for S3, flink-hadoop-fs for HDFS).
3. Test FileSystem.get(basePath.toUri()).createRecoverableWriter() independently to get the root cause IOException.
4. Check filesystem configuration in flink-conf.yaml (fs.s3a.endpoint, fs.defaultFS, etc.).
5. Verify credentials and permissions for the target filesystem.

// Validate filesystem accessibility before building the sink
Path basePath = new Path("hdfs://namenode:8020/output");
FileSystem fs = FileSystem.get(basePath.toUri());
if (!fs.exists(basePath.getParent())) {
    throw new IOException("Base path parent does not exist: " + basePath);
}
fs.createRecoverableWriter(); // should not throw

try {
    sink.getWriterStateSerializer();
} catch (FlinkRuntimeException e) {
    if (e.getMessage().equals("Could not create writer state serializer.")) {
        // inspect e.getCause() (IOException) for filesystem root cause
    }
    throw e;
}

When it happens

Trigger: The FileSink's basePath points to a filesystem (HDFS, S3, local) that is unreachable or misconfigured; the filesystem scheme is not registered; the path URI is malformed; the required filesystem plugin JAR is missing from the classpath.

Common situations: S3 endpoint misconfigured or credentials missing; HDFS not running; missing flink-s3-fs-hadoop or flink-s3-fs-native plugin JAR; wrong path scheme (e.g., s3a:// vs s3://); filesystem configuration not set in flink-conf.yaml.

Related errors

• Could not create committable serializer.
• At least one trigger condition must be configured.
• The given configuration directory name '{}' ({}) does not de
• The Flink config file '{}' ({}) does not exist.
• Can not create a Path from an empty string

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