apache/flink · error · IOException

Stream is closed

Error message

Stream is closed

What it means

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.

Source

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

    private void createNewTempFile() throws IOException {
        File tmpDir = new File(localTmpDir);
        Files.createDirectories(tmpDir.toPath());

        currentTempFile = new File(tmpDir, "s3-part-" + UUID.randomUUID());
        currentFileStream = new FileOutputStream(currentTempFile);
        currentOutputStream = new BufferedOutputStream(currentFileStream, BUFFER_SIZE);
        currentPartSize = 0;
    }

    @Override
    public long getPos() throws IOException {
        return numBytesInParts + currentPartSize;
    }

    @Override
    public void write(int b) throws IOException {
        if (closed) {
            throw new IOException("Stream is closed");
        }

        currentOutputStream.write(b);
        currentPartSize++;

        if (currentPartSize >= minPartSize) {
            uploadCurrentPart();
            createNewTempFile();
        }
    }

    @Override
    public void write(byte[] b, int off, int len) throws IOException {
        if (closed) {
            throw new IOException("Stream is closed");
        }
        if (b == null) {
            throw new NullPointerException();

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Audit call order in your sink/output format: ensure no write happens after close()/closeForCommit().
2. Guard writes with the stream's own state or track a local 'closed' boolean in the caller and skip writes after it.
3. If you need to append after commit, open a NEW stream via the RecoverableWriter instead of reusing the closed one.
4. For multi-threaded access, serialize writes and close behind one lock.

Example fix

// before
stream.write(b); // may run after close()
stream.close();

// after
if (!streamClosed) {
    stream.write(b);
}
// ... later
stream.close();
streamClosed = true;

// FSDataOutputStream exposes no isOpen(); track in caller
if (!streamClosed) {
    stream.write(b);
}

When it happens

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

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

Related errors

• Stream is already 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/227e692b92fdf1a3. Report an issue: GitHub.