What causes "This consumer has already been closed." in apache/kafka?

Ensure no thread can call the consumer after close(): use a volatile closed flag or an AtomicReference swapped to null on close.

apache/kafka · error · java.lang.IllegalStateException

This consumer has already been closed.

Error message

This consumer has already been closed.

What it means

Thrown by acquireAndEnsureOpen() — the gatekeeper invoked by every public state-mutating consumer method — when this.closed is true. Once KafkaConsumer.close() has run, the instance is permanently unusable; any subsequent call (poll, subscribe, assign, commit, etc.) trips this IllegalStateException. The release() is called before throwing so the internal lock is not leaked.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:2207

    @Override
    public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) {
        if (listener == null)
            throw new IllegalArgumentException("RebalanceListener cannot be null");

        subscribeInternal(pattern, Optional.of(listener));
    }

    /**
     * Acquire the light lock and ensure that the consumer hasn't been closed.
     *
     * @throws IllegalStateException If the consumer has been closed
     */
    private void acquireAndEnsureOpen() {
        acquire();
        if (this.closed) {
            release();
            throw new IllegalStateException("This consumer has already been closed.");
        }

        try {
            metadata.maybeThrowBootstrapFatalException();
        } catch (RuntimeException e) {
            release();
            throw e;
        }
    }

    /**
     * Acquire the light lock protecting this consumer from multithreaded access. Instead of blocking
     * when the lock is not available, however, we just throw an exception (since multithreaded usage is not
     * supported).
     *
     * @throws ConcurrentModificationException if another thread already has the lock
     */
    private void acquire() {

View on GitHub (pinned to c31c9215e1)

Solutions

Ensure no thread can call the consumer after close(): use a volatile closed flag or an AtomicReference<KafkaConsumer> swapped to null on close.
For thread handoff, publish the consumer via a thread-safe holder and have the consumer thread check isAlive / closed before each poll.
In Spring, scope the consumer bean to the listener container's lifecycle so @PreDestroy and the consumer thread cannot race.
If using try-with-resources, perform all consumer operations inside the block.

Example fix

// before
KafkaConsumer<String,String> c = new KafkaConsumer<>(props);
// ... later, c.close() called by shutdown hook
c.poll(Duration.ofMillis(100)); // throws

// after
private final AtomicReference<KafkaConsumer<String,String>> ref = new AtomicReference<>(c);
// shutdown hook:
KafkaConsumer<String,String> c = ref.getAndSet(null);
if (c != null) c.close();
// consumer thread:
KafkaConsumer<String,String> cur = ref.get();
if (cur == null) return; // already closed
try { cur.poll(Duration.ofMillis(100)); }
catch (IllegalStateException e) { /* expected during shutdown */ }

Defensive patterns

Strategy: try-catch

Validate before calling

// Track closed state alongside the consumer in your wrapper:
private volatile boolean closed = false;
public synchronized void safeClose() {
    if (closed) return;
    closed = true;
    consumer.close();
}
// Before any consumer call:
if (closed) throw new IllegalStateException("consumer closed");

Type guard

// Cannot introspect KafkaConsumer#closed from outside; track locally.
// Use a wrapper exposing isClosed():
final class SafeConsumer<K,V> {
    private final KafkaConsumer<K,V> delegate;
    private volatile boolean closed = false;
    boolean isClosed() { return closed; }
}

Try / catch

try {
    consumer.poll(Duration.ofMillis(100));
} catch (IllegalStateException e) {
    if (e.getMessage().contains("already been closed")) {
        log.info("Consumer closed; skipping further poll");
        return; // or recreate the consumer
    }
    throw e;
}

Prevention

Own the consumer lifecycle in one place (e.g. a try-with-resources or a managed bean).
After close(), null out the reference and guard all access with a null check or isClosed() flag.
Never share the consumer across components with independent close() calls; centralize shutdown.
If you recreate consumers, use a wrapper that swaps the underlying instance atomically and exposes isClosed().

When it happens

Trigger: Calling any consumer method after close() returned; calling poll() in a loop after another thread or a shutdown hook closed the consumer; using a consumer inside a try-with-resources block and then accessing it after the block; Spring @PreDestroy closing the consumer before an in-flight task finished.

Common situations: Graceful shutdown hooks racing with worker threads; connection pools/wrappers that close consumers on error and then retry on the same instance; test fixtures that close consumers in @AfterEach but a background thread still references them; reactive frameworks cancelling a subscription that triggers close while a poll is queued.

Related errors

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