apache/pulsar · warning · ServerMetadataException

Consumer was not connected

Error message

Consumer was not connected

What it means

AbstractDispatcherSingleActiveConsumer.removeConsumer throws ServerMetadataException 'Consumer was not connected' when the consumer being removed is not present in the dispatcher's consumers list. The close/disconnect path expected this consumer to be registered on this dispatcher, so removal indicates a state mismatch between the consumer's ownership and the dispatcher's subscription state.

Source

Thrown at pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java:265

        consumers.add(consumer);

        if (!pickAndScheduleActiveConsumer()) {
            // the active consumer is not changed
            Consumer currentActiveConsumer = getActiveConsumer();
            if (null == currentActiveConsumer) {
                log.debug().attr("consumer", consumer).log("Current active consumer disappears while adding consumer");
            } else {
                consumer.notifyActiveConsumerChange(currentActiveConsumer);
            }
        }

        return CompletableFuture.completedFuture(null);
    }

    public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
        log.info().attr("consumer", consumer).log("Removing consumer");
        if (!consumers.remove(consumer)) {
            throw new ServerMetadataException("Consumer was not connected");
        }

        if (consumers.isEmpty()) {
            activeConsumer = null;
        }

        if (closeFuture == null && !consumers.isEmpty()) {
            pickAndScheduleActiveConsumer();
            return;
        }

        cancelPendingRead();

        if (consumers.isEmpty() && closeFuture != null && !closeFuture.isDone()) {
            // Control reaches here only when closeFuture is created
            // and no more connected consumers left.
            closeFuture.complete(null);
        }

View on GitHub (pinned to 820761864e)

Solutions

1. On the client, avoid double-closing consumers (guard close() with an idempotent flag)
2. Verify the client is connected to the broker that owns the consumer (check lookup/connection logs)
3. Retry subscribe/consumer creation after this error — the dispatcher state has self-corrected (consumer absent)
4. If seen frequently with failover, check broker version for known dispatcher race fixes and upgrade

Example fix

// before
consumer.close();
consumer.close(); // second close triggers server-side error
// after
if (consumer != null && !closed) {
    consumer.close();
    closed = true;
}

// client: ensure the consumer is open and connected before close
if (consumer == null || !consumer.isConnected()) return;

boolean safeToClose(Consumer<?> c) {
    return c != null && c.isConnected() && !c.getLastDisconnectedTimestamp().isPresent() == false;
}

// server-side throw is ServerMetadataException; clients see it as broker-side error on close
try {
    consumer.close();
} catch (PulsarClientException e) {
    log.warn("close failed; consumer may have already been removed", e); // idempotent close
}

When it happens

Trigger: Calling removeConsumer with a Consumer instance not in the dispatcher's list — consumer already removed by a concurrent close, consumer connected to a different broker than where close is attempted, or a race where the consumer disconnected and re-registered elsewhere first.

Common situations: Client closing a consumer twice or after a reconnect moved it to another broker; failover where the broker still holds a stale consumer reference; race conditions during subscription unload.

Related errors

• The topic has a max partition index of %d, the number of par
• entryFilterNames can't be empty. To remove entry filters use
• The offloadPolicies must be specified for namespace offload.
• The driver is not supported, support value: ${supportedDrive
• The bucket must be specified for namespace offload.

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