What causes "The stream is already closed" in apache/kafka?

Audit call sites to guarantee no write() occurs after close(); reorder so close() is the last operation on the stream.

apache/kafka · error · IllegalStateException

The stream is already closed

Error message

The stream is already closed

What it means

Thrown as IllegalStateException by ensureNotFinished() in Lz4BlockOutputStream whenever write(int) or write(byte[],int,int) is invoked after the stream has been closed (the finished flag was set to true in close()). The LZ4 frame has already had its EndMark written; any further write would produce an invalid frame, so the class rejects it.

Source

Thrown at clients/src/main/java/org/apache/kafka/common/compress/Lz4BlockOutputStream.java:233

        bufferOffset += len;
    }

    @Override
    public void flush() throws IOException {
        if (!finished) {
            writeBlock();
        }
        if (out != null) {
            out.flush();
        }
    }

    /**
     * A simple state check to ensure the stream is still open.
     */
    private void ensureNotFinished() {
        if (finished) {
            throw new IllegalStateException(CLOSED_STREAM);
        }
    }

    @Override
    public void close() throws IOException {
        try {
            if (!finished) {
                // basically flush the buffer writing the last block
                writeBlock();
                // write the end block
                writeEndMark();
            }
        } finally {
            try {
                if (out != null) {
                    try (OutputStream outStream = out) {
                        outStream.flush();
                    }

View on GitHub (pinned to c31c9215e1)

Solutions

Audit call sites to guarantee no write() occurs after close(); reorder so close() is the last operation on the stream.
If a wrapper may legitimately receive late writes, guard with a boolean closed flag and drop or buffer them rather than passing through.
Avoid reusing the Lz4BlockOutputStream instance across produce calls — create a fresh one per batch.
Reproduce with assertions enabled and a stack trace to find the stray post-close write.

Example fix

// before: cleanup closes the stream, then a callback writes to it
try (Lz4BlockOutputStream out = new Lz4BlockOutputStream(ch, level, broken)) {
    out.write(payload);
}
listener.onAck(() -> out.write(trailer)); // IllegalStateException

// after: complete all writes before close, or guard late writes
try (Lz4BlockOutputStream out = new Lz4BlockOutputStream(ch, level, broken)) {
    out.write(payload);
    out.write(trailer);
}

Defensive patterns

Strategy: validation

Validate before calling

// Lz4BlockOutputStream flips to 'finished' after close(); writes then throw.
// Track lifecycle externally and check before every write.
Lz4BlockOutputStream out = ...;
if (!isOpen(out)) {                       // see typeGuard
    throw new IllegalStateException("Attempted write to already-closed LZ4 stream");
}
out.write(b, off, len);

Type guard

// There is no public isOpen() on Lz4BlockOutputStream; mirror state in a holder.
class Lz4Sink {
    private final Lz4BlockOutputStream out;
    private boolean open = true;
    boolean isOpen() { return open; }
    void close() throws IOException { try { out.close(); } finally { open = false; } }
}

Try / catch

try {
    out.write(b, off, len);
} catch (IllegalStateException e) {
    if (Lz4BlockOutputStream.CLOSED_STREAM.equals(e.getMessage())) {
        // discard the write target and re-create a fresh LZ4 frame if needed
        reopenSink();
    } else throw e;
}

Prevention

Use try-with-resources around Lz4BlockOutputStream so close() is invoked exactly once and the lifecycle is lexical.
Never share a single Lz4BlockOutputStream across threads or producer callbacks; the class is documented non-thread-safe and 'finished' is global state.
Set out = null immediately after close() at every ownership boundary and null-check before write.
If wrapping in an appender/pipeline, expose an explicit isOpen() in your wrapper rather than relying on the underlying class.

When it happens

Trigger: Calling write()/write(int)/write(byte[],int,int) on a Lz4BlockOutputStream after close() has run (close() sets finished=true at line 257). Reached via ensureNotFinished() at lines 189 and 199.

Common situations: Try-with-resources where a callback or async path writes after the block exited; pooled/reused OutputStream wrappers that don't track lifecycle; cleanup code that closes the stream defensively and then a finally/interceptor writes a trailer; integration code that wraps a Kafka producer's compression stream and outlives the produce call; bugs in flush-then-close ordering in custom serializers.

Related errors

AI-assisted analysis of apache/kafka@c31c9215e1 (2026-08-03). Data as JSON: /data/errors/d5b39c79feb79419.json. Report an issue: GitHub.