{"record":{"id":"69cf82fbeb496690","repo":"apache/flink","slug":"could-not-create-committable-serializer","errorCode":null,"errorMessage":"Could not create committable serializer.","messagePattern":"Could not create committable serializer\\.","errorType":"exception","errorClass":"FlinkRuntimeException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java","lineNumber":188,"sourceCode":"            throw new FlinkRuntimeException(\"Could not create writer state serializer.\", e);\n        }\n    }\n\n    @Override\n    public Committer<FileSinkCommittable> createCommitter(CommitterInitContext context)\n            throws IOException {\n        return bucketsBuilder.createCommitter();\n    }\n\n    @Override\n    public SimpleVersionedSerializer<FileSinkCommittable> getCommittableSerializer() {\n        try {\n            return bucketsBuilder.getCommittableSerializer();\n        } catch (IOException e) {\n            // it's not optimal that we have to do this but creating the serializers for the\n            // FileSink requires (among other things) a call to FileSystem.get() which declares\n            // IOException.\n            throw new FlinkRuntimeException(\"Could not create committable serializer.\", e);\n        }\n    }\n\n    @Override\n    public SimpleVersionedSerializer<FileSinkCommittable> getWriteResultSerializer() {\n        return getCommittableSerializer();\n    }\n\n    @Override\n    public Collection<String> getCompatibleWriterStateNames() {\n        // StreamingFileSink\n        return Collections.singleton(\"bucket-states\");\n    }\n\n    public static <IN> DefaultRowFormatBuilder<IN> forRowFormat(\n            final Path basePath, final Encoder<IN> encoder) {\n        return new DefaultRowFormatBuilder<>(basePath, encoder, new DateTimeBucketAssigner<>());\n    }","sourceCodeStart":170,"sourceCodeEnd":206,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java#L170-L206","documentation":"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.","triggerScenarios":"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.","commonSituations":"S3 endpoint misconfigured or credentials missing; HDFS not running; missing filesystem plugin JAR; wrong path scheme; filesystem configuration not set in flink-conf.yaml.","solutions":["Verify the basePath URI is correct and the target filesystem is accessible.","Ensure the required filesystem plugin JAR is on the classpath.","Test FileSystem.get(basePath.toUri()).createRecoverableWriter() independently to isolate the root cause.","Check filesystem configuration and credentials in flink-conf.yaml."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Validate filesystem before building the sink\nPath basePath = new Path(\"s3a://my-bucket/output\");\nFileSystem fs = FileSystem.get(basePath.toUri());\nfs.createRecoverableWriter(); // should not throw","typeGuard":null,"tryCatchPattern":"try {\n    sink.getCommittableSerializer();\n} catch (FlinkRuntimeException e) {\n    if (e.getMessage().equals(\"Could not create committable serializer.\")) {\n        // inspect e.getCause() (IOException) for filesystem root cause\n    }\n    throw e;\n}","preventionTips":["Verify the filesystem is accessible and the recoverable writer can be created.","Ensure the filesystem plugin JAR is on the classpath.","Test filesystem connectivity before deploying the job."],"tags":["file-sink","filesystem","serializer","configuration","connectivity"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}