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

Audit the lifecycle: ensure the poll/acknowledge loop is stopped before consumer.close() is invoked.

apache/kafka · error · IllegalStateException

This consumer has already been closed.

Error message

This consumer has already been closed.

What it means

Thrown by acquireAndEnsureOpen() (line 1127) when this.closed is true. Almost every public ShareConsumer method funnels through acquireAndEnsureOpen, so once the consumer is closed any further API call (poll, acknowledge, subscribe, commitSync, metrics, etc.) is rejected immediately rather than operating on released resources. This preserves the lifecycle invariant that a closed consumer cannot be reused.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:1127

    /**
     * {@inheritDoc}
     */
    @Override
    public void wakeup() {
        wakeupTrigger.wakeup();
    }

    /**
     * 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() {
        final Thread thread = Thread.currentThread();

View on GitHub (pinned to c31c9215e1)

Solutions

Audit the lifecycle: ensure the poll/acknowledge loop is stopped before consumer.close() is invoked.
Do not reuse a closed consumer — create a new KafkaShareConsumer instance instead.
If using try-with-resources, make sure no asynchronous task (scheduler, callback, listener) outlives the try block.
Add an isOpen()/closed check in your own loop, or cancel the loop's executor before closing.

Example fix

// before
try (var consumer = new KafkaShareConsumer<>(props)) {
    scheduler.scheduleAtFixedRate(() -> consumer.poll(Duration.ofMillis(500)), 0, 1, TimeUnit.SECONDS);
}
// scheduler fires after close -> exception

// after
ScheduledFuture<?> task = scheduler.scheduleAtFixedRate(pollLoop, 0, 1, TimeUnit.SECONDS);
try (var consumer = new KafkaShareConsumer<>(props)) {
    // ... use consumer ...
} finally {
    task.cancel(false);
}

Defensive patterns

Strategy: validation

Validate before calling

// Track the consumer lifecycle in a wrapper so callers cannot touch a closed consumer.
private volatile boolean closed = false;
private final org.apache.kafka.clients.consumer.ShareConsumer<K,V> delegate;

public void safePoll(long ms) {
    if (closed) throw new IllegalStateException("share consumer already closed");
    delegate.poll(java.time.Duration.ofMillis(ms));
}

public void close() {
    closed = true;
    delegate.close();
}

Type guard

null

Try / catch

try {
    consumer.poll(java.time.Duration.ofMillis(500));
} catch (IllegalStateException e) {
    if (e.getMessage() != null && e.getMessage().contains("already been closed")) {
        // consumer is unusable; mark it as dead and recreate if needed
        log.warn("Share consumer was closed; skipping further use", e);
    } else {
        throw e;
    }
}

Prevention

Prefer try-with-resources (ShareConsumer extends AutoCloseable) to make closure explicit and deterministic.
After calling close(), null out the reference so stale holders cannot invoke it again.
In multi-owner code, centralize ownership so only one component owns the lifecycle.

When it happens

Trigger: Calling any public method on a KafkaShareConsumer after close() has returned (or after a try-with-resources block has auto-closed it). Also reached when a background poll loop continues one iteration after the owning component closed the consumer, or when a callback fires post-close.

Common situations: try-with-resources scoping error where the consumer is closed at the end of the block but a captured lambda/scheduled task still references it; shutdown ordering in Spring/Quarkus where @PreDestroy on the consumer fires before the poll-loop bean is stopped; handing the consumer to a library that retains it beyond its intended lifetime; double-closing in a finally block after an earlier close.

Related errors

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