apache/flink · error · FlinkRuntimeException

Could not create committable serializer.

Error message

Could not create committable serializer.

What it means

Thrown by getCommittableSerializer() in FileSink when bucketsBuilder.getCommittableSerializer() fails with IOException. Like the writer state serializer path, the underlying createBucketWriter() triggers FileSystem.get(basePath.toUri()).createRecoverableWriter(), which can throw IOException. Wrapped as FlinkRuntimeException because the Sink interface method does not declare checked exceptions.

Source

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

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

    @Override
    public SimpleVersionedSerializer<FileSinkCommittable> getWriteResultSerializer() {
        return getCommittableSerializer();
    }

    @Override
    public Collection<String> getCompatibleWriterStateNames() {
        // StreamingFileSink
        return Collections.singleton("bucket-states");
    }

    public static <IN> DefaultRowFormatBuilder<IN> forRowFormat(
            final Path basePath, final Encoder<IN> encoder) {
        return new DefaultRowFormatBuilder<>(basePath, encoder, new DateTimeBucketAssigner<>());
    }

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 JAR is on the classpath.
3. Test FileSystem.get(basePath.toUri()).createRecoverableWriter() independently to isolate the root cause.
4. Check filesystem configuration and credentials in flink-conf.yaml.

// Validate filesystem before building the sink
Path basePath = new Path("s3a://my-bucket/output");
FileSystem fs = FileSystem.get(basePath.toUri());
fs.createRecoverableWriter(); // should not throw

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

When it happens

Trigger: Same root causes as the writer state serializer failure: the basePath filesystem is unreachable, misconfigured, or the filesystem plugin is missing. This method is called during sink initialization to create the committable serialization infrastructure.

Common situations: S3 endpoint misconfigured or credentials missing; HDFS not running; missing filesystem plugin JAR; wrong path scheme; filesystem configuration not set in flink-conf.yaml.

Related errors

• Could not create writer state 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/69cf82fbeb496690. Report an issue: GitHub.