apache/flink · error · IOException

Stream is already closed

Error message

Stream is already closed

What it means

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.

Source

Thrown at flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java:219

        // unmasked and let close() perform cleanup. nextPartNumber is only advanced on success so a
        // failed attempt does not leave a gap in the part sequence.
        NativeS3ObjectOperations.UploadPartResult result =
                s3AccessHelper.uploadPart(
                        key, uploadId, nextPartNumber, currentTempFile, currentPartSize);

        nextPartNumber++;
        completedParts.add(new PartETag(result.getPartNumber(), result.getETag()));
        numBytesInParts += currentPartSize;

        Files.delete(currentTempFile.toPath());
    }

    @Override
    public Committer closeForCommit() throws IOException {
        lock();
        try {
            if (closed) {
                throw new IOException("Stream is already closed");
            }

            currentOutputStream.close();

            if (currentPartSize > 0) {
                uploadCurrentPart();
            } else {
                Files.delete(currentTempFile.toPath());
            }

            NativeS3Recoverable recoverable =
                    new NativeS3Recoverable(
                            key, uploadId, new ArrayList<>(completedParts), numBytesInParts);

            closed = true;
            return new NativeS3Committer(s3AccessHelper, recoverable);
        } finally {
            unlock();

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Commit exactly once: null the stream reference immediately after closeForCommit() succeeds.
2. 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.
3. Wrap the commit in a try-finally that only closes (never re-commits) on error.
4. Add an integration test that asserts single-commit semantics under failure injection.

Example fix

// before
try { committer = stream.closeForCommit(); }
catch (IOException e) { committer = stream.closeForCommit(); } // throws 'already closed'

// after
Committer committer = stream.closeForCommit();
stream = null; // make double-commit impossible to write by accident

if (stream != null) {
    committer = stream.closeForCommit();
    stream = null;
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Stream is closed
• RecoverableWriter has been closed
• Unable to write to output stream.
• Premature EOF from inputStream
• Premeture EOF from inputStream

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/30d18649721ac39b. Report an issue: GitHub.