{"record":{"id":"30d18649721ac39b","repo":"apache/flink","slug":"stream-is-already-closed","errorCode":null,"errorMessage":"Stream is already closed","messagePattern":"Stream is already 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":219,"sourceCode":"        // unmasked and let close() perform cleanup. nextPartNumber is only advanced on success so a\n        // failed attempt does not leave a gap in the part sequence.\n        NativeS3ObjectOperations.UploadPartResult result =\n                s3AccessHelper.uploadPart(\n                        key, uploadId, nextPartNumber, currentTempFile, currentPartSize);\n\n        nextPartNumber++;\n        completedParts.add(new PartETag(result.getPartNumber(), result.getETag()));\n        numBytesInParts += currentPartSize;\n\n        Files.delete(currentTempFile.toPath());\n    }\n\n    @Override\n    public Committer closeForCommit() throws IOException {\n        lock();\n        try {\n            if (closed) {\n                throw new IOException(\"Stream is already closed\");\n            }\n\n            currentOutputStream.close();\n\n            if (currentPartSize > 0) {\n                uploadCurrentPart();\n            } else {\n                Files.delete(currentTempFile.toPath());\n            }\n\n            NativeS3Recoverable recoverable =\n                    new NativeS3Recoverable(\n                            key, uploadId, new ArrayList<>(completedParts), numBytesInParts);\n\n            closed = true;\n            return new NativeS3Committer(s3AccessHelper, recoverable);\n        } finally {\n            unlock();","sourceCodeStart":201,"sourceCodeEnd":237,"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#L201-L237","documentation":"closeForCommit() throws IOException(\"Stream is already closed\") when invoked a second time or after close(). Committing is terminal: it uploads the final part, builds the NativeS3Recoverable, and marks the stream closed, so a second commit attempt is a caller bug.","triggerScenarios":"Calling closeForCommit() twice on the same NativeS3RecoverableFsDataOutputStream, or calling it after close(); commonly happens in retry logic around commit, or when both snapshotState and a finally block try to commit.","commonSituations":"Sink code that retries the commit block after a downstream exception without realizing the stream already committed; two code paths (failure handler + normal path) both committing; recovering and committing the same underlying stream instance.","solutions":["Commit exactly once: null the stream reference immediately after closeForCommit() succeeds.","On failure during commit, do not blindly re-call closeForCommit() — the failure may have occurred after the stream closed; use recover-for-commit via the Recoverable instead.","Wrap the commit in a try-finally that only closes (never re-commits) on error.","Add an integration test that asserts single-commit semantics under failure injection."],"exampleFix":"// before\ntry { committer = stream.closeForCommit(); }\ncatch (IOException e) { committer = stream.closeForCommit(); } // throws 'already closed'\n\n// after\nCommitter committer = stream.closeForCommit();\nstream = null; // make double-commit impossible to write by accident","handlingStrategy":"validation","validationCode":"if (stream != null) {\n    committer = stream.closeForCommit();\n    stream = null;\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["One closeForCommit per stream — enforce by nulling the reference.","On commit failure, use the recoverable-for-commit path, never a second closeForCommit.","Avoid try/catch-retry wrappers around commit."],"tags":["s3","stream","commit","lifecycle","io"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}