apache/pulsar · error · PulsarClientException.AlreadyClosedException

Topic was terminated

Error message

Topic was terminated

What it means

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.

Source

Thrown at pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java:1033

        if (batchReceivePolicy.getMaxNumMessages() <= 0 && batchReceivePolicy.getMaxNumBytes() <= 0) {
            return false;
        }
        return (batchReceivePolicy.getMaxNumMessages() > 0
                && incomingMessages.size() >= batchReceivePolicy.getMaxNumMessages())
                || (batchReceivePolicy.getMaxNumBytes() > 0
                && getIncomingMessageSize() >= batchReceivePolicy.getMaxNumBytes());
    }

    private void verifyConsumerState() throws PulsarClientException {
        switch (getState()) {
            case Ready:
            case Connecting:
                break; // Ok
            case Closing:
            case Closed:
                throw  new PulsarClientException.AlreadyClosedException("Consumer already closed");
            case Terminated:
                throw new PulsarClientException.AlreadyClosedException("Topic was terminated");
            case Failed:
            case Uninitialized:
                throw new PulsarClientException.NotConnectedException();
            default:
                break;
        }
    }

    private void verifyBatchReceive() throws PulsarClientException {
        if (listener != null) {
            throw new PulsarClientException.InvalidConfigurationException(
                "Cannot use receive() when a listener has been set");
        }
        if (getCurrentReceiverQueueSize() == 0) {
            throw new PulsarClientException.InvalidConfigurationException(
                "Can't use batch receive, if the queue size is 0");
        }
    }

View on GitHub (pinned to 820761864e)

Solutions

1. 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).
2. Use ConsumerBuilder#subscribe to a non-terminated topic, or re-enable the topic by deleting and recreating it if the terminated state is unintended.
3. If a reader/consumer pattern is used for finite data, catch AlreadyClosedException and treat it as normal end-of-stream.

Example fix

// before
while (true) { Message<byte[]> m = consumer.receive(); process(m); }
// after
try {
  while (true) { Message<byte[]> m = consumer.receive(); process(m); }
} catch (PulsarClientException.AlreadyClosedException e) {
  // topic terminated: end of finite stream
}

// No public API to check termination pre-call; treat AlreadyClosedException as end-of-stream signal.

try {
  Message<byte[]> m = consumer.receive(1, TimeUnit.SECONDS);
  if (m != null) process(m);
} catch (PulsarClientException.AlreadyClosedException e) {
  // topic terminated: stop polling / recreate consumer
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Consumer already closed
• Consumer was not connected
• Cannot use receive() when a listener has been set
• Can't use receive with timeout, if the queue size is 0
• Can't use batch receive, if the queue size is 0

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/afafe67b3956c11d. Report an issue: GitHub.