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

Create a new KafkaConsumer instance after closing; do not reuse a closed consumer.

apache/kafka · error · IllegalStateException

This consumer has already been closed.

Error message

This consumer has already been closed.

What it means

Thrown by acquireAndEnsureOpen() whenever any consumer operation is invoked after close() has set the closed flag. The consumer is single-use for open operations, so once closed it rejects all public API calls with this IllegalStateException. It protects against use-after-close in shared or pooled consumers.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java:1237

        // are partitions with a missing position, then we will raise an exception.
        subscriptions.resetInitializingPositions();

        // Finally send an asynchronous request to look up and update the positions of any
        // partitions which are awaiting reset.
        offsetFetcher.resetPositionsIfNeeded();

        return true;
    }

    /**
     * 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 multi-threaded access. Instead of blocking
     * when the lock is not available, however, we just throw an exception (since multi-threaded usage is not
     * supported).
     * @throws ConcurrentModificationException if another thread already has the lock
     */
    private void acquire() {
        final Thread thread = Thread.currentThread();
        final long threadId = thread.getId();

View on GitHub (pinned to c31c9215e1)

Solutions

Create a new KafkaConsumer instance after closing; do not reuse a closed consumer.
Synchronize close and usage so no operation runs after close; use a single owner thread for the consumer lifecycle.
Guard pooled consumers with an 'open' flag and re-instantiate on demand instead of reopening.
Fix lifecycle wiring in DI frameworks to ensure tasks stop before @PreDestroy.

Example fix

// before
consumer.close();
consumer.poll(Duration.ofMillis(100)); // throws

// after
consumer.close();
consumer = new KafkaConsumer<>(props); // new instance
consumer.poll(Duration.ofMillis(100));

Defensive patterns

Strategy: type-guard

Validate before calling

// Track lifecycle yourself so you never call a method on a closed consumer:
private volatile boolean closed = false;
public synchronized void close() {
    if (closed) return;
    closed = true;
    consumer.close();
}
public long position(TopicPartition tp) {
    if (closed) throw new IllegalStateException("consumer already closed");
    return consumer.position(tp);
}

Type guard

// A wrapper type whose state machine makes post-close calls unrepresentable:
final class ManagedConsumer<K,V> implements AutoCloseable {
    private KafkaConsumer<K,V> consumer; // null once closed
    public synchronized long position(TopicPartition tp) {
        KafkaConsumer<K,V> c = consumer;
        if (c == null) throw new IllegalStateException("closed");
        return c.position(tp);
    }
    public synchronized void close() {
        if (consumer != null) { consumer.close(); consumer = null; }
    }
}

Try / catch

// If you receive this from an opaque consumer handle, downgrade to a no-op for the closed case:
try {
    consumer.position(tp);
} catch (IllegalStateException e) {
    if (e.getMessage().contains("already been closed")) {
        log.debug("Consumer closed; ignoring position() call");
        return -1L;
    }
    throw e;
}

Prevention

Use try-with-resources on the consumer (it's AutoCloseable) so close happens exactly once at a well-defined scope.
Never share a consumer across threads or re-enter it from a callback that fires after close(); the closed flag is set inside close().
In async pipelines, gate every consumer interaction behind an AtomicBoolean isOpen so a race between shutdown and a worker doesn't hit this error.

When it happens

Trigger: Calling poll(), commitSync(), position(), subscribe(), etc. after consumer.close(); a shutdown thread closing the consumer while another thread still calls it; dependency injection scopes returning a closed bean; try-with-resources then reusing the consumer.

Common situations: Multi-threaded code where a background thread closes the consumer; Spring/lifecycle beans where @PreDestroy fired but a scheduled task still runs; try-with-resources blocks leaking references; frameworks (Camel, Spring Cloud Stream) mismanaging consumer lifecycle on restart.

Related errors

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