{"record":{"id":"f15fc0fabff07908","repo":"apache/kafka","slug":"this-consumer-has-already-been-closed-f15fc0","errorCode":null,"errorMessage":"This consumer has already been closed.","messagePattern":"This consumer has already been closed\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"clients/src/main/java/org/apache/kafka/clients/consumer/MockShareConsumer.java","lineNumber":191,"sourceCode":"    }\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":173,"sourceCodeEnd":193,"githubUrl":"https://github.com/apache/kafka/blob/996fb4585aa1bcc8980b0e1b8d6b168b986cd979/clients/src/main/java/org/apache/kafka/clients/consumer/MockShareConsumer.java#L173-L193","documentation":"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.","triggerScenarios":"Calling addRecord (or any guarded method such as metrics()) on a MockShareConsumer after close() has been invoked.","commonSituations":"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.","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."],"exampleFix":"// before\nMockShareConsumer<K,V> c = new MockShareConsumer<>();\nc.close();\nc.addRecord(record); // -> IllegalStateException\n\n// after\nMockShareConsumer<K,V> c = new MockShareConsumer<>();\nc.addRecord(record);\nc.poll(Duration.ZERO);\nc.close();","handlingStrategy":"validation","validationCode":"null  // no public isOpen() on the mock; track close() in the test","typeGuard":"null","tryCatchPattern":"try {\n    mock.addRecord(record);\n} catch (IllegalStateException e) {\n    if (e.getMessage().contains(\"already been closed\")) { /* re-create or skip */ }\n    else throw e;\n}","preventionTips":["Create a fresh MockShareConsumer per test rather than reusing across tests.","In @AfterEach, only close() - never call addRecord/poll."],"tags":["testing","mock","share-consumer","lifecycle","kafka-clients"],"backgroundTag":null,"analyzedSha":"996fb4585aa1bcc8980b0e1b8d6b168b986cd979","analyzedAt":"2026-08-11T22:03:28.655Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}