apache/kafka · error · IllegalStateException

Cannot add records for a partition that is not assigned to t

Error message

Cannot add records for a partition that is not assigned to the consumer

What it means

Thrown by MockConsumer.addRecord when the record's topic-partition is not in the consumer's current assignment. MockConsumer only buffers records for partitions the mock has assigned (mirroring real consumer semantics where you can only consume what you are assigned). Adding a record to an unassigned partition would never be polled and indicates a test-setup bug.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java:383

                }
            }
        }

        return new ConsumerRecords<>(results, nextOffsetAndMetadata);
    }

    /**
     * Adds a record to be returned when {@link #poll(Duration)} is called.
     *
     * @param record the record to add
     * @throws IllegalStateException if the partition is not assigned to the consumer
     */
    public synchronized void addRecord(ConsumerRecord<K, V> record) {
        ensureNotClosed();
        TopicPartition tp = new TopicPartition(record.topic(), record.partition());
        Set<TopicPartition> currentAssigned = this.subscriptions.assignedPartitions();
        if (!currentAssigned.contains(tp))
            throw new IllegalStateException("Cannot add records for a partition that is not assigned to the consumer");
        List<ConsumerRecord<K, V>> recs = records.computeIfAbsent(tp, k -> new ArrayList<>());
        recs.add(record);
    }

    /**
     * Sets the maximum number of records returned in a single call to {@link #poll(Duration)}.
     *
     * @param maxPollRecords the max.poll.records.
     */
    public synchronized void setMaxPollRecords(long maxPollRecords) {
        if (maxPollRecords < 1) {
            throw new IllegalArgumentException("MaxPollRecords must be strictly superior to 0");
        }
        this.maxPollRecords = maxPollRecords;
    }

    /**
     * Sets an exception to throw when {@link #poll(Duration)} is called.

View on GitHub (pinned to 996fb4585a)

Solutions

Call mock.assign(Set.of(new TopicPartition(record.topic(), record.partition()))) before addRecord, or trigger a rebalance so the partition is assigned.
Derive the TopicPartition from the same constant used in assign() to avoid name/number drift.
Assert mock.assignment().contains(tp) before addRecord in a test helper.

Example fix

// before
ConsumerRecord<String,String> rec = new ConsumerRecord<>("orders", 0, 0L, "k", "v");
mockConsumer.addRecord(rec); // partition 0 not assigned

// after
TopicPartition tp = new TopicPartition("orders", 0);
mockConsumer.assign(Set.of(tp));
mockConsumer.addRecord(rec);

Defensive patterns

Strategy: validation

Validate before calling

TopicPartition tp = new TopicPartition(record.topic(), record.partition());
if (!mockConsumer.assignment().contains(tp))
    mockConsumer.assign(union(mockConsumer.assignment(), Set.of(tp)));
mockConsumer.addRecord(record);

Type guard

static boolean isAssigned(MockConsumer<?,?> m, TopicPartition tp) {
    return m.assignment().contains(tp);
}

Try / catch

try {
    mockConsumer.addRecord(record);
} catch (IllegalStateException e) {
    if ("Cannot add records for a partition that is not assigned to the consumer".equals(e.getMessage())) {
        TopicPartition tp = new TopicPartition(record.topic(), record.partition());
        mockConsumer.assign(union(mockConsumer.assignment(), Set.of(tp)));
        mockConsumer.addRecord(record);
    } else throw e;
}

Prevention

Always assign partitions in the test before adding records.
Use a helper that ensures the partition is assigned before addRecord.
Derive TopicPartition references from constants shared between assign and addRecord.

When it happens

Trigger: Calling addRecord before assign()/rebalance; record.partition() not matching any assigned TopicPartition; topic name mismatch between assign and addRecord; assigned tp0 but adding to tp1.

Common situations: Tests that subscribe (not assign) and forget to call rebalance() to populate the assignment; refactor that changed topic names in one place but not the other; copy-paste tests with stale partition numbers.

Related errors

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