{"record":{"id":"1e15c193f0578b8e","repo":"apache/flink","slug":"could-not-create-writer-state-serializer","errorCode":null,"errorMessage":"Could not create writer state serializer.","messagePattern":"Could not create writer state 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":170,"sourceCode":"\n    @Override\n    public FileWriter<IN> restoreWriter(\n            WriterInitContext context, Collection<FileWriterBucketState> recoveredState)\n            throws IOException {\n        FileWriter<IN> writer = bucketsBuilder.createWriter(context);\n        writer.initializeState(recoveredState);\n        return writer;\n    }\n\n    @Override\n    public SimpleVersionedSerializer<FileWriterBucketState> getWriterStateSerializer() {\n        try {\n            return bucketsBuilder.getWriterStateSerializer();\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 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);","sourceCodeStart":152,"sourceCodeEnd":188,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java#L152-L188","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Verify the basePath URI is correct and the target filesystem is accessible.","Ensure the required filesystem plugin is on the classpath (e.g., flink-s3-fs-hadoop for S3, flink-hadoop-fs for HDFS).","Test FileSystem.get(basePath.toUri()).createRecoverableWriter() independently to get the root cause IOException.","Check filesystem configuration in flink-conf.yaml (fs.s3a.endpoint, fs.defaultFS, etc.).","Verify credentials and permissions for the target filesystem."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Validate filesystem accessibility before building the sink\nPath basePath = new Path(\"hdfs://namenode:8020/output\");\nFileSystem fs = FileSystem.get(basePath.toUri());\nif (!fs.exists(basePath.getParent())) {\n    throw new IOException(\"Base path parent does not exist: \" + basePath);\n}\nfs.createRecoverableWriter(); // should not throw","typeGuard":null,"tryCatchPattern":"try {\n    sink.getWriterStateSerializer();\n} catch (FlinkRuntimeException e) {\n    if (e.getMessage().equals(\"Could not create writer state serializer.\")) {\n        // inspect e.getCause() (IOException) for filesystem root cause\n    }\n    throw e;\n}","preventionTips":["Verify the filesystem scheme is registered and the plugin JAR is on the classpath.","Test FileSystem.get(path).createRecoverableWriter() before building the sink.","Ensure credentials and endpoints are configured in flink-conf.yaml."],"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"}