{"record":{"id":"afafe67b3956c11d","repo":"apache/pulsar","slug":"topic-was-terminated","errorCode":null,"errorMessage":"Topic was terminated","messagePattern":"Topic was terminated","errorType":"exception","errorClass":"PulsarClientException.AlreadyClosedException","httpStatus":null,"severity":"error","filePath":"pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java","lineNumber":1033,"sourceCode":"        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\");\n        }\n    }","sourceCodeStart":1015,"sourceCodeEnd":1051,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java#L1015-L1051","documentation":"Pulsar throws this as an AlreadyClosedException when the consumer's state is Terminated, meaning the topic the consumer is attached to was explicitly terminated (a durable 'no more messages' marker set via the admin API). After termination no further receives can succeed, so the consumer fails fast instead of blocking.","triggerScenarios":"Calling receive(), receiveAsync(), batchReceive() (or nextMessage-style paths that call checkClientState) while the consumer's internal state is Terminated, i.e. the broker sent a TopicTerminatedError to this consumer.","commonSituations":"Consuming from a topic with retention/termination enabled in a finite-stream workload; someone ran `pulsar-admin topics terminate`; a recreated topic was terminated; tests that terminate topics then keep polling.","solutions":["Stop consuming from this topic and create a new consumer on the new/recreated topic (terminating resets it only when the topic is deleted and recreated).","Use ConsumerBuilder#subscribe to a non-terminated topic, or re-enable the topic by deleting and recreating it if the terminated state is unintended.","If a reader/consumer pattern is used for finite data, catch AlreadyClosedException and treat it as normal end-of-stream."],"exampleFix":"// before\nwhile (true) { Message<byte[]> m = consumer.receive(); process(m); }\n// after\ntry {\n  while (true) { Message<byte[]> m = consumer.receive(); process(m); }\n} catch (PulsarClientException.AlreadyClosedException e) {\n  // topic terminated: end of finite stream\n}","handlingStrategy":"try-catch","validationCode":"// No public API to check termination pre-call; treat AlreadyClosedException as end-of-stream signal.","typeGuard":null,"tryCatchPattern":"try {\n  Message<byte[]> m = consumer.receive(1, TimeUnit.SECONDS);\n  if (m != null) process(m);\n} catch (PulsarClientException.AlreadyClosedException e) {\n  // topic terminated: stop polling / recreate consumer\n}","preventionTips":["Avoid terminating topics that active consumers still poll; prefer deletion+recreation with a documented cutover.","Use receive(timeout) so the loop can observe state changes gracefully.","Subscribe with the latest position on terminated finite streams to treat termination as completion."],"tags":["pulsar","topic-terminated","consumer","lifecycle"],"backgroundTag":"topic-terminated","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"}