{"record":{"id":"d2f171f349807892","repo":"apache/pulsar","slug":"consumer-already-closed","errorCode":null,"errorMessage":"Consumer already closed","messagePattern":"Consumer already closed","errorType":"exception","errorClass":"PulsarClientException.AlreadyClosedException","httpStatus":null,"severity":"error","filePath":"pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java","lineNumber":1031,"sourceCode":"\n    protected boolean hasEnoughMessagesForBatchReceive() {\n        if (batchReceivePolicy.getMaxNumMessages() <= 0 && batchReceivePolicy.getMaxNumBytes() <= 0) {\n            return false;\n        }\n        return (batchReceivePolicy.getMaxNumMessages() > 0\n                && incomingMessages.size() >= batchReceivePolicy.getMaxNumMessages())\n                || (batchReceivePolicy.getMaxNumBytes() > 0\n                && getIncomingMessageSize() >= batchReceivePolicy.getMaxNumBytes());\n    }\n\n    private void verifyConsumerState() throws PulsarClientException {\n        switch (getState()) {\n            case Ready:\n            case Connecting:\n                break; // Ok\n            case Closing:\n            case Closed:\n                throw  new PulsarClientException.AlreadyClosedException(\"Consumer already closed\");\n            case Terminated:\n                throw new PulsarClientException.AlreadyClosedException(\"Topic was terminated\");\n            case Failed:\n            case Uninitialized:\n                throw new PulsarClientException.NotConnectedException();\n            default:\n                break;\n        }\n    }\n\n    private void verifyBatchReceive() throws PulsarClientException {\n        if (listener != null) {\n            throw new PulsarClientException.InvalidConfigurationException(\n                \"Cannot use receive() when a listener has been set\");\n        }\n        if (getCurrentReceiverQueueSize() == 0) {\n            throw new PulsarClientException.InvalidConfigurationException(\n                \"Can't use batch receive, if the queue size is 0\");","sourceCodeStart":1013,"sourceCodeEnd":1049,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java#L1013-L1049","documentation":"ConsumerBase.verifyConsumerState() (state machine switch on getState()) throws AlreadyClosedException 'Consumer already closed' when the consumer is in Closing or Closed state and any operation (receive, ack, etc.) is attempted. After close()/asyncClose() completes, the consumer is unusable; this exception signals lifecycle misuse rather than a connectivity problem.","triggerScenarios":"Calling any consumer method after consumer.close() (or after the consumer was auto-closed via try-with-resources exiting), or racing close() with in-flight receive/ack calls so the op lands during Closing state.","commonSituations":"Using a consumer after a try-with-resources block returns it; application shutdown hook closing consumers while worker threads still poll; caching consumers in a registry where an admin path closed one but producers still reference it; Pulsar client state change from another thread.","solutions":["Check consumer's state (via getLastDisconnectedTimestamp or wrapping calls) or catch AlreadyClosedException and re-obtain a consumer","Do not share consumers across lifecycles — create a new consumer after close","Synchronize shutdown: stop producer threads before closing consumers","Keep consumers long-lived; Pulsar consumers are designed to be reused, not opened/closed per message"],"exampleFix":"// before\ntry (Consumer<String> c = buildConsumer()) {\n    process(c);\n}\nc.receive(); // AlreadyClosedException\n// after\nConsumer<String> c = buildConsumer();\ntry {\n    process(c);\n} finally {\n    c.close();\n}","handlingStrategy":"try-catch","validationCode":"// track consumer lifecycle yourself; do not use after close()\nif (closed) {\n    throw new IllegalStateException(\"Consumer already closed; recreate before use\");\n}","typeGuard":"boolean isUsable(ConsumerBase<?> c) {\n    var state = c.getState();\n    return state == HandlerState.State.Ready || state == HandlerState.State.Connecting;\n}","tryCatchPattern":"try {\n    consumer.receive();\n} catch (PulsarClientException.AlreadyClosedException e) {\n    consumer = recreateConsumer(); // rebuild after lifecycle misuse\n}","preventionTips":["Keep consumers long-lived; do not close per message or per request","Stop worker threads before closing consumers in shutdown hooks","Never return a try-with-resources consumer to calling code after the block exits","Catch AlreadyClosedException at the consumer-access layer and transparently recreate"],"tags":["pulsar","consumer","lifecycle","already-closed"],"backgroundTag":"consumer-already-closed","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}