{"record":{"id":"cfc4b97a4d0efc62","repo":"apache/kafka","slug":"cannot-add-records-for-a-topics-that-is-not-subscr","errorCode":null,"errorMessage":"Cannot add records for a topics that is not subscribed by the consumer","messagePattern":"Cannot add records for a topics that is not subscribed by the consumer","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"clients/src/main/java/org/apache/kafka/clients/consumer/MockShareConsumer.java","lineNumber":184,"sourceCode":"    public synchronized void close() {\n        close(Duration.ofMillis(DEFAULT_CLOSE_TIMEOUT_MS));\n    }\n\n    @Override\n    public synchronized void close(Duration timeout) {\n        closed = true;\n    }\n\n    @Override\n    public synchronized void wakeup() {\n        wakeup.set(true);\n    }\n\n    public synchronized void addRecord(ConsumerRecord<K, V> record) {\n        ensureNotClosed();\n        TopicPartition tp = new TopicPartition(record.topic(), record.partition());\n        if (!subscriptions.subscription().contains(record.topic()))\n            throw new IllegalStateException(\"Cannot add records for a topics that is not subscribed by the consumer\");\n        List<ConsumerRecord<K, V>> recs = records.computeIfAbsent(tp, k -> new ArrayList<>());\n        recs.add(record);\n    }\n\n    private void ensureNotClosed() {\n        if (closed)\n            throw new IllegalStateException(\"This consumer has already been closed.\");\n    }\n}","sourceCodeStart":166,"sourceCodeEnd":193,"githubUrl":"https://github.com/apache/kafka/blob/996fb4585aa1bcc8980b0e1b8d6b168b986cd979/clients/src/main/java/org/apache/kafka/clients/consumer/MockShareConsumer.java#L166-L193","documentation":"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.)","triggerScenarios":"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.","commonSituations":"Test setup ordering bug - adding fixture records before subscribe(); wrong topic name in the fixture; or the test unsubscribed between adding records.","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."],"exampleFix":"// before\nMockShareConsumer<String,String> c = new MockShareConsumer<>();\nc.addRecord(new ConsumerRecord<>(\"t\", 0, 0, \"k\", \"v\")); // -> IllegalStateException\n\n// after\nMockShareConsumer<String,String> c = new MockShareConsumer<>();\nc.subscribe(Collections.singleton(\"t\"));\nc.addRecord(new ConsumerRecord<>(\"t\", 0, 0, \"k\", \"v\"));","handlingStrategy":"validation","validationCode":"// Only addRecord for topics we have subscribed to.\nString topic = record.topic();\nif (!mockShareConsumer.subscription().contains(topic)) {\n    mockShareConsumer.subscribe(Set.of(topic));\n}\nmockShareConsumer.addRecord(record);","typeGuard":"null","tryCatchPattern":"null","preventionTips":["Subscribe to all fixture topics once during setup.","Use a helper addRecord(consumer, record) that auto-subscribes if needed."],"tags":["testing","mock","share-consumer","subscription","kafka-clients"],"backgroundTag":null,"analyzedSha":"996fb4585aa1bcc8980b0e1b8d6b168b986cd979","analyzedAt":"2026-08-11T22:03:28.655Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}