apache/pulsar · error · PulsarClientException.AlreadyClosedException

Consumer already closed

Error message

Consumer already closed

What it means

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.

Source

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

    protected boolean hasEnoughMessagesForBatchReceive() {
        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. Check consumer's state (via getLastDisconnectedTimestamp or wrapping calls) or catch AlreadyClosedException and re-obtain a consumer
2. Do not share consumers across lifecycles — create a new consumer after close
3. Synchronize shutdown: stop producer threads before closing consumers
4. Keep consumers long-lived; Pulsar consumers are designed to be reused, not opened/closed per message

Example fix

// before
try (Consumer<String> c = buildConsumer()) {
    process(c);
}
c.receive(); // AlreadyClosedException
// after
Consumer<String> c = buildConsumer();
try {
    process(c);
} finally {
    c.close();
}

// track consumer lifecycle yourself; do not use after close()
if (closed) {
    throw new IllegalStateException("Consumer already closed; recreate before use");
}

boolean isUsable(ConsumerBase<?> c) {
    var state = c.getState();
    return state == HandlerState.State.Ready || state == HandlerState.State.Connecting;
}

try {
    consumer.receive();
} catch (PulsarClientException.AlreadyClosedException e) {
    consumer = recreateConsumer(); // rebuild after lifecycle misuse
}

When it happens

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

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

Related errors

• Topic was terminated
• 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/d2f171f349807892. Report an issue: GitHub.