What causes "Producer closed while allocating memory" in apache/kafka?

Order shutdown: stop accepting new sends first (signal workers to drain), await in-flight sends/callbacks, then close the producer. Use producer.close(Duration) with a timeout to flush.

apache/kafka · error · KafkaException

Producer closed while allocating memory

Error message

Producer closed while allocating memory

What it means

Thrown by BufferPool.allocate when the producer's BufferPool has already been closed by the time a thread acquires the lock to allocate. Closing the producer marks the pool closed; any subsequent or in-flight allocation attempt that reaches the allocate path aborts with a KafkaException rather than handing out memory from a defunct pool.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java:151

     * @throws IllegalArgumentException if size is larger than the total memory controlled by the pool (and hence we would block
     *         forever)
     */
    public ByteBuffer allocate(int size, long maxTimeToBlockMs) throws InterruptedException {
        if (allocationMode != AllocationMode.FULL)
            throw new IllegalStateException("allocate() is not supported in " + allocationMode
                + " allocation mode; use allocateChunks()");
        if (size > this.totalMemory)
            throw new IllegalArgumentException("Attempt to allocate " + size
                                               + " bytes, but there is a hard limit of "
                                               + this.totalMemory
                                               + " on memory allocations.");

        ByteBuffer buffer = null;
        this.lock.lock();

        if (this.closed) {
            this.lock.unlock();
            throw new KafkaException("Producer closed while allocating memory");
        }

        try {
            // check if we have a free buffer of the right size pooled
            if (size == poolableSize && !this.free.isEmpty())
                return this.free.pollFirst();

            // now check if the request is immediately satisfiable with the
            // memory on hand or if we need to block
            int freeListSize = freeSize() * this.poolableSize;
            if (this.nonPooledAvailableMemory + freeListSize >= size) {
                // we have enough unallocated or pooled memory to immediately
                // satisfy the request, but need to allocate the buffer
                freeUp(size);
                this.nonPooledAvailableMemory -= size;
            } else {
                // we are out of memory and will have to block
                int accumulated = 0;

View on GitHub (pinned to c31c9215e1)

Solutions

Order shutdown: stop accepting new sends first (signal workers to drain), await in-flight sends/callbacks, then close the producer. Use producer.close(Duration) with a timeout to flush.
Guard application code so sends are not issued after close: track a 'closed' flag in your producer wrapper and reject new sends with a clear domain exception before they hit the buffer pool.
Use a single owner for the producer's lifecycle (e.g. a managed singleton) and ensure no thread can reach producer.send after the owner has begun closing it.

Example fix

// before - shutdown hook closes producer while workers still sending
Runtime.getRuntime().addShutdownHook(new Thread(producer::close));

// after - drain workers first, then close with a flush timeout
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    workerPool.shutdown();
    workerPool.awaitTermination(30, TimeUnit.SECONDS);
    producer.close(Duration.ofSeconds(10));
}));

Defensive patterns

Strategy: try-catch

Try / catch

// Producer was closed before/at allocation. send() surfaces this as KafkaException.
try {
    producer.send(record);
} catch (org.apache.kafka.common.KafkaException e) {
    if (e.getMessage() != null && e.getMessage().contains("Producer closed")) {
        // producer is shutting down; stop submitting, route remaining records elsewhere
    } else throw e;
}

Prevention

Close the producer only after all futures returned by send() have completed (callback or .get()).
Use a single owner thread for producer lifecycle; never share close() with send() threads unsynchronized.
Track an AtomicBoolean 'closed' in your wrapper and check it before submitting, as a fast-fail hint.
In shutdown hooks, flush() then close() with a timeout, and stop enqueuing new records first.

When it happens

Trigger: One thread calls KafkaProducer.close() while another thread is concurrently calling producer.send(...) whose internal allocate() runs after close set the closed flag. Also when a send is attempted after the producer was already closed elsewhere (e.g. shutdown hook, container teardown, or a finally block that closed the producer before a still-running worker finished).

Common situations: Application shutdown races where the producer bean/resource is closed before all worker threads have drained; try-with-resources or @PreDestroy closing the producer while background senders are still active; integration tests that close the producer between iterations but reuse worker threads; reconnect/restart logic that closes a failed producer while callbacks are still in flight.

Related errors

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