apache/kafka · error · KafkaException

Producer closed while send in progress

Error message

Producer closed while send in progress

What it means

Thrown by ChunkedRecordAccumulator.tryAppend when the accumulator has been closed (the producer is shutting down) at the moment a send thread attempts to append to an existing batch. It is the incremental-strategy analogue of the BufferPool close checks: the producer's chunked accumulator refuses to route a record into a batch once close has been called, surfacing a KafkaException back through producer.send.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulator.java:322

    /**
     * Try to append to a ProducerBatch, with mid-batch chunk extension support.
     * <p>
     * If the open batch is within its batch-size limit but its chunked stream lacks chunk
     * capacity, returns {@link RecordAppendResult#needsExtension(int)} without
     * attempting the append; the caller allocates chunks outside the deque lock, attaches
     * them, and retries. Otherwise defers to the parent, which appends or returns
     * {@link RecordAppendResult#NEEDS_NEW_BATCH}.
     *
     * @return a {@link RecordAppendResult#needsExtension(int) needsBufferExtension} result when the open batch is
     * within its batch-size limit but its chunks lack capacity (the append is not attempted); otherwise the
     * parent implementation's outcome: an {@code appended} result ({@link RecordAppendResult#appended()}) or
     * {@link RecordAppendResult#NEEDS_NEW_BATCH}.
     */
    @Override
    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();
        // Split batches in an incremental deque are plain ProducerBatch (heap-backed, grow-on-demand)
        // and never need chunk extension, so the check only applies to chunked batches.
        if (last instanceof ChunkedProducerBatch) {
            int extensionBytes = ((ChunkedProducerBatch) last).extensionBytesNeeded(timestamp, key, value, headers);
            if (extensionBytes > 0)
                return RecordAppendResult.needsExtension(extensionBytes);
        }
        return super.tryAppend(timestamp, key, value, headers, callback, deque, nowMs);
    }

    @Override
    protected ProducerBatch createProducerBatch(TopicPartition tp, MemoryRecordsBuilder recordsBuilder, long nowMs) {
        return new ChunkedProducerBatch(tp, recordsBuilder, nowMs);
    }

    /**
     * Build a {@link MemoryRecordsBuilder} backed by the chunked stream.

View on GitHub (pinned to c31c9215e1)

Solutions

1. Drive shutdown through a single owner: set a 'closing' flag, stop submitting new sends, await in-flight sends/callbacks, then producer.close(Duration).
2. Wrap producer.send in a component that tracks open/closed state and rejects sends with a clear domain exception once close has begun, so callers never see the raw KafkaException.
3. Ensure background tasks (schedulers, callbacks) are cancelled before the producer is closed, e.g. via executor.shutdown() + awaitTermination before producer.close.

Example fix

// before - scheduler outlives the producer bean
scheduler.submit(() -> producer.send(rec));
producer.close(); // scheduler may still call send -> tryAppend throws

// after - cancel scheduler, then close
scheduler.shutdown();
scheduler.awaitTermination(30, TimeUnit.SECONDS);
producer.close(Duration.ofSeconds(10));

// tryAppend throws KafkaException when the accumulator is closed mid-send.
try {
    producer.send(record);
} catch (org.apache.kafka.common.KafkaException e) {
    if (e.getMessage() != null && e.getMessage().contains("Producer closed")) {
        // The chunked accumulator saw close() during the append critical section.
        // Treat as terminal for this producer; re-enqueue elsewhere.
    } else throw e;
}

When it happens

Trigger: Calling producer.send(...) on a thread after another thread has closed the producer (or set the accumulator's closed flag via close). The send reaches ChunkedRecordAccumulator.append, which calls tryAppend on the current partition's deque; tryAppend checks this.closed and throws before touching any batch.

Common situations: Shutdown races where worker threads keep calling send while a coordinator/container/@PreDestroy hook closes the producer; failed-producer recreation logic that closes the old producer before all in-flight sends have completed; background scheduler tasks that outlive the producer bean in a Spring/CDI context; tests reusing a producer instance across methods and closing it early.

Related errors

• Producer closed while allocating memory
• Attempt to allocate {totalSize} bytes ({numChunks} chunks of
• Compression is not yet supported with the incremental buffer
• Requested metadata update after close
• Producer closed while send in progress

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