{"record":{"id":"736703d013c7089f","repo":"apache/kafka","slug":"cannot-add-records-for-a-partition-that-is-not-ass","errorCode":null,"errorMessage":"Cannot add records for a partition that is not assigned to the consumer","messagePattern":"Cannot add records for a partition that is not assigned to the consumer","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java","lineNumber":383,"sourceCode":"                }\n            }\n        }\n\n        return new ConsumerRecords<>(results, nextOffsetAndMetadata);\n    }\n\n    /**\n     * Adds a record to be returned when {@link #poll(Duration)} is called.\n     *\n     * @param record the record to add\n     * @throws IllegalStateException if the partition is not assigned to the consumer\n     */\n    public synchronized void addRecord(ConsumerRecord<K, V> record) {\n        ensureNotClosed();\n        TopicPartition tp = new TopicPartition(record.topic(), record.partition());\n        Set<TopicPartition> currentAssigned = this.subscriptions.assignedPartitions();\n        if (!currentAssigned.contains(tp))\n            throw new IllegalStateException(\"Cannot add records for a partition that is not assigned to the consumer\");\n        List<ConsumerRecord<K, V>> recs = records.computeIfAbsent(tp, k -> new ArrayList<>());\n        recs.add(record);\n    }\n\n    /**\n     * Sets the maximum number of records returned in a single call to {@link #poll(Duration)}.\n     *\n     * @param maxPollRecords the max.poll.records.\n     */\n    public synchronized void setMaxPollRecords(long maxPollRecords) {\n        if (maxPollRecords < 1) {\n            throw new IllegalArgumentException(\"MaxPollRecords must be strictly superior to 0\");\n        }\n        this.maxPollRecords = maxPollRecords;\n    }\n\n    /**\n     * Sets an exception to throw when {@link #poll(Duration)} is called.","sourceCodeStart":365,"sourceCodeEnd":401,"githubUrl":"https://github.com/apache/kafka/blob/996fb4585aa1bcc8980b0e1b8d6b168b986cd979/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java#L365-L401","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before\nConsumerRecord<String,String> rec = new ConsumerRecord<>(\"orders\", 0, 0L, \"k\", \"v\");\nmockConsumer.addRecord(rec); // partition 0 not assigned\n\n// after\nTopicPartition tp = new TopicPartition(\"orders\", 0);\nmockConsumer.assign(Set.of(tp));\nmockConsumer.addRecord(rec);","handlingStrategy":"validation","validationCode":"TopicPartition tp = new TopicPartition(record.topic(), record.partition());\nif (!mockConsumer.assignment().contains(tp))\n    mockConsumer.assign(union(mockConsumer.assignment(), Set.of(tp)));\nmockConsumer.addRecord(record);","typeGuard":"static boolean isAssigned(MockConsumer<?,?> m, TopicPartition tp) {\n    return m.assignment().contains(tp);\n}","tryCatchPattern":"try {\n    mockConsumer.addRecord(record);\n} catch (IllegalStateException e) {\n    if (\"Cannot add records for a partition that is not assigned to the consumer\".equals(e.getMessage())) {\n        TopicPartition tp = new TopicPartition(record.topic(), record.partition());\n        mockConsumer.assign(union(mockConsumer.assignment(), Set.of(tp)));\n        mockConsumer.addRecord(record);\n    } else throw e;\n}","preventionTips":["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."],"tags":["consumer","mock-consumer","test","rebalance","illegal-state"],"backgroundTag":null,"analyzedSha":"996fb4585aa1bcc8980b0e1b8d6b168b986cd979","analyzedAt":"2026-08-11T22:03:28.655Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}