apache/kafka · error · IllegalStateException

Cannot add records for a topics that is not subscribed by th

Error message

Cannot add records for a topics that is not subscribed by the consumer

What it means

MockShareConsumer.addRecord enforces that records can only be queued for topics the consumer is currently subscribed to, by checking subscriptions.subscription().contains(record.topic()). If addRecord is called before subscribe (or for a topic that was never subscribed), it throws IllegalStateException. (Note the upstream message has a typo 'a topics'; that is the literal string shipped by the library.)

Source

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

    public synchronized void close() {
        close(Duration.ofMillis(DEFAULT_CLOSE_TIMEOUT_MS));
    }

    @Override
    public synchronized void close(Duration timeout) {
        closed = true;
    }

    @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

Call mockShareConsumer.subscribe(Collections.singleton(record.topic())) before addRecord.
Verify the record's topic name matches the subscribed topic exactly (case-sensitive).
If you intend to feed records for multiple topics, subscribe to a collection containing all of them first.

Example fix

// before
MockShareConsumer<String,String> c = new MockShareConsumer<>();
c.addRecord(new ConsumerRecord<>("t", 0, 0, "k", "v")); // -> IllegalStateException

// after
MockShareConsumer<String,String> c = new MockShareConsumer<>();
c.subscribe(Collections.singleton("t"));
c.addRecord(new ConsumerRecord<>("t", 0, 0, "k", "v"));

Defensive patterns

Strategy: validation

Validate before calling

// Only addRecord for topics we have subscribed to.
String topic = record.topic();
if (!mockShareConsumer.subscription().contains(topic)) {
    mockShareConsumer.subscribe(Set.of(topic));
}
mockShareConsumer.addRecord(record);

Type guard

null

Try / catch

null

Prevention

Subscribe to all fixture topics once during setup.
Use a helper addRecord(consumer, record) that auto-subscribes if needed.

When it happens

Trigger: Calling mockShareConsumer.addRecord(record) where record.topic() is not in the current subscription set - either because subscribe() was never called, was called with a different topic, or was unsubscribed.

Common situations: Test setup ordering bug - adding fixture records before subscribe(); wrong topic name in the fixture; or the test unsubscribed between adding records.

Related errors

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