What causes "Producer closed while send in progress" in apache/kafka?

Stop all send()-calling code before invoking close(); gate sends with an AtomicBoolean 'running' flag flipped in shutdown.

apache/kafka · critical · KafkaException

Producer closed while send in progress

Error message

Producer closed while send in progress

What it means

Thrown by RecordAccumulator.tryAppend when the accumulator's `closed` flag is true at the moment a record is being appended. tryAppend runs on the user thread inside KafkaProducer.send(); a true flag means close() has already been called (or is in progress), so any further append is rejected to avoid orphaned records that can never be drained by the Sender thread.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:473

    /**
     * Try to append to a ProducerBatch.
     * <p>
     * If it is full (or absent), we return {@link RecordAppendResult#NEEDS_NEW_BATCH} and a new batch is created.
     * We also close the batch for record appends to free up resources like compression buffers. The batch will be
     * fully closed (ie. the record batch headers will be written and memory records built) in one of the following
     * cases (whichever comes first): right before send, if it is expired, or when the producer is closed.
     *
     * @return one of two outcomes: an {@code appended} result ({@link RecordAppendResult#appended()}) when the
     * record was appended to the open batch, or {@link RecordAppendResult#NEEDS_NEW_BATCH} when there is
     * no open batch that can take it (full or absent). The incremental strategy overrides this and may
     * additionally return a {@link RecordAppendResult#needsExtension(int) needsBufferExtension} result
     * when the open batch is within its batch-size limit but its chunks lack capacity for the record.
     */
    protected RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers,
                                           Callback callback, Deque<ProducerBatch> deque, long nowMs) {
        if (closed)
            throw new KafkaException("Producer closed while send in progress");
        ProducerBatch last = deque.peekLast();
        if (last != null) {
            int initialBytes = last.estimatedSizeInBytes();
            FutureRecordMetadata future = last.tryAppend(timestamp, key, value, headers, callback, nowMs);
            if (future == null) {
                last.closeForRecordAppends();
            } else {
                int appendedBytes = last.estimatedSizeInBytes() - initialBytes;
                return RecordAppendResult.appended(future, deque.size() > 1 || last.isFull(), false, appendedBytes);
            }
        }
        return RecordAppendResult.NEEDS_NEW_BATCH;
    }

    private boolean isMuted(TopicPartition tp) {
        return muted.contains(tp);
    }

View on GitHub (pinned to c31c9215e1)

Solutions

Stop all send()-calling code before invoking close(); gate sends with an AtomicBoolean 'running' flag flipped in shutdown.
Use a single owner for the producer lifecycle (one creator, one closer) and never share it across independently-stopped components.
Catch this KafkaException at the call site and treat the record as not-sent, then route to a fallback or DLQ; do not retry on the same closed instance.
After close(), create a fresh KafkaProducer if production must continue rather than reusing the closed one.

Example fix

// before
Runtime.getRuntime().addShutdownHook(new Thread(producer::close));
// worker threads keep calling producer.send() during shutdown

// after
private final AtomicBoolean open = new AtomicBoolean(true);
// in worker:
if (!open.get()) throw new IllegalStateException("producer closing");
producer.send(rec);
// in shutdown hook:
open.set(false);
producer.close(Duration.ofSeconds(10));

Defensive patterns

Strategy: try-catch

Validate before calling

// Coordinate close vs send via an AtomicBoolean to eliminate the race at the call site.
private final java.util.concurrent.atomic.AtomicBoolean closed = new java.util.concurrent.atomic.AtomicBoolean();
...
if (closed.get()) return; // drop the send; producer is shutting down
producer.send(record, callback);

Try / catch

try {
    producer.send(record, callback);
} catch (org.apache.kafka.common.KafkaException ke) {
    if (ke.getMessage() != null && ke.getMessage().contains("Producer closed")) {
        // close() raced this send; record was not accepted, safe to drop or hand off
    }
}

Prevention

Guarantee a happens-before edge between close() and the last send(): close after all senders have stopped.
Use a shutdown flag checked by the sender thread before each send(); close() only after the sender exits.
Never share a producer across threads where one thread may close while another is still sending.

When it happens

Trigger: Calling KafkaProducer.send() after producer.close() has begun; or close() racing with a concurrent send(). The check at the top of tryAppend fails immediately and propagates out of send() to the caller.

Common situations: Producer reused after shutdown in a long-running service; shutdown hook closing the producer while worker threads still enqueue; Spring bean destroyed but background jobs still reference it; a producer field set to null on error and a later send() on a stale reference whose close() was triggered.

Related errors

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