{"record":{"id":"227e692b92fdf1a3","repo":"apache/flink","slug":"stream-is-closed-227e69","errorCode":null,"errorMessage":"Stream is closed","messagePattern":"Stream is closed","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java","lineNumber":148,"sourceCode":"    private void createNewTempFile() throws IOException {\n        File tmpDir = new File(localTmpDir);\n        Files.createDirectories(tmpDir.toPath());\n\n        currentTempFile = new File(tmpDir, \"s3-part-\" + UUID.randomUUID());\n        currentFileStream = new FileOutputStream(currentTempFile);\n        currentOutputStream = new BufferedOutputStream(currentFileStream, BUFFER_SIZE);\n        currentPartSize = 0;\n    }\n\n    @Override\n    public long getPos() throws IOException {\n        return numBytesInParts + currentPartSize;\n    }\n\n    @Override\n    public void write(int b) throws IOException {\n        if (closed) {\n            throw new IOException(\"Stream is closed\");\n        }\n\n        currentOutputStream.write(b);\n        currentPartSize++;\n\n        if (currentPartSize >= minPartSize) {\n            uploadCurrentPart();\n            createNewTempFile();\n        }\n    }\n\n    @Override\n    public void write(byte[] b, int off, int len) throws IOException {\n        if (closed) {\n            throw new IOException(\"Stream is closed\");\n        }\n        if (b == null) {\n            throw new NullPointerException();","sourceCodeStart":130,"sourceCodeEnd":166,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java#L130-L166","documentation":"The single-byte write(int b) rejects calls after the stream's closed flag was set. Like most Flink FSDataOutputStream implementations, this stream is single-use: once close()/closeForCommit() ran, further writes throw IOException(\"Stream is closed\") rather than silently dropping bytes.","triggerScenarios":"Calling write(int) after close() or closeForCommit() — typically a user function or sink that keeps a stale stream reference, a close-then-flush race in operator lifecycle methods, or an output format that writes a trailer during cleanup after already closing the stream.","commonSituations":"Sink operator reusing a recovered output stream after commit; close() invoked in a finally block that also writes a final record; two threads sharing one RecoverableFsDataOutputStream; recover() returning a new stream while old code paths still write to the old instance.","solutions":["Audit call order in your sink/output format: ensure no write happens after close()/closeForCommit().","Guard writes with the stream's own state or track a local 'closed' boolean in the caller and skip writes after it.","If you need to append after commit, open a NEW stream via the RecoverableWriter instead of reusing the closed one.","For multi-threaded access, serialize writes and close behind one lock."],"exampleFix":"// before\nstream.write(b); // may run after close()\nstream.close();\n\n// after\nif (!streamClosed) {\n    stream.write(b);\n}\n// ... later\nstream.close();\nstreamClosed = true;","handlingStrategy":"validation","validationCode":"// FSDataOutputStream exposes no isOpen(); track in caller\nif (!streamClosed) {\n    stream.write(b);\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Null the stream reference right after close().","Single-writer ownership per stream; no shared references across threads.","Order sink code so cleanup never writes."],"tags":["s3","stream","lifecycle","io"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}