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

Reorder teardown so all addRecord/poll calls happen before close().

apache/kafka · error · IllegalStateException

This consumer has already been closed.

Error message

This consumer has already been closed.

What it means

MockShareConsumer tracks a 'closed' flag set by close(). Every public mutative method calls ensureNotClosed() which throws IllegalStateException('This consumer has already been closed.') once the flag is true. This mirrors the real consumer's post-close rejection so tests catch use-after-close bugs.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/MockShareConsumer.java:191

    }

    @Override
    public synchronized void wakeup() {
        wakeup.set(true);
    }

    public synchronized void addRecord(ConsumerRecord<K, V> record) {
        ensureNotClosed();
        TopicPartition tp = new TopicPartition(record.topic(), record.partition());
        if (!subscriptions.subscription().contains(record.topic()))
            throw new IllegalStateException("Cannot add records for a topics that is not subscribed by the consumer");
        List<ConsumerRecord<K, V>> recs = records.computeIfAbsent(tp, k -> new ArrayList<>());
        recs.add(record);
    }

    private void ensureNotClosed() {
        if (closed)
            throw new IllegalStateException("This consumer has already been closed.");
    }
}

View on GitHub (pinned to 996fb4585a)

Solutions

Reorder teardown so all addRecord/poll calls happen before close().
Create a fresh MockShareConsumer per test rather than reusing a closed instance.
If you need to assert post-close behavior, catch IllegalStateException explicitly instead of letting it fail the test.

Example fix

// before
MockShareConsumer<K,V> c = new MockShareConsumer<>();
c.close();
c.addRecord(record); // -> IllegalStateException

// after
MockShareConsumer<K,V> c = new MockShareConsumer<>();
c.addRecord(record);
c.poll(Duration.ZERO);
c.close();

Defensive patterns

Strategy: validation

Validate before calling

null  // no public isOpen() on the mock; track close() in the test

Type guard

null

Try / catch

try {
    mock.addRecord(record);
} catch (IllegalStateException e) {
    if (e.getMessage().contains("already been closed")) { /* re-create or skip */ }
    else throw e;
}

Prevention

Create a fresh MockShareConsumer per test rather than reusing across tests.
In @AfterEach, only close() - never call addRecord/poll.

When it happens

Trigger: Calling addRecord (or any guarded method such as metrics()) on a MockShareConsumer after close() has been invoked.

Common situations: Test @AfterEach ordering that closes the consumer before late-arriving fixture additions; polling/cleanup running after an explicit close; reusing a shared mock across test methods without re-instantiating.

Related errors

AI-assisted analysis of apache/kafka@996fb4585a (2026-08-11). Data as JSON: /api/errors/f15fc0fabff07908. Report an issue: GitHub.