{"record":{"id":"47acc5aa22022678","repo":"apache/pulsar","slug":"consumer-was-not-connected","errorCode":null,"errorMessage":"Consumer was not connected","messagePattern":"Consumer was not connected","errorType":"exception","errorClass":"ServerMetadataException","httpStatus":null,"severity":"warning","filePath":"pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java","lineNumber":265,"sourceCode":"        consumers.add(consumer);\n\n        if (!pickAndScheduleActiveConsumer()) {\n            // the active consumer is not changed\n            Consumer currentActiveConsumer = getActiveConsumer();\n            if (null == currentActiveConsumer) {\n                log.debug().attr(\"consumer\", consumer).log(\"Current active consumer disappears while adding consumer\");\n            } else {\n                consumer.notifyActiveConsumerChange(currentActiveConsumer);\n            }\n        }\n\n        return CompletableFuture.completedFuture(null);\n    }\n\n    public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {\n        log.info().attr(\"consumer\", consumer).log(\"Removing consumer\");\n        if (!consumers.remove(consumer)) {\n            throw new ServerMetadataException(\"Consumer was not connected\");\n        }\n\n        if (consumers.isEmpty()) {\n            activeConsumer = null;\n        }\n\n        if (closeFuture == null && !consumers.isEmpty()) {\n            pickAndScheduleActiveConsumer();\n            return;\n        }\n\n        cancelPendingRead();\n\n        if (consumers.isEmpty() && closeFuture != null && !closeFuture.isDone()) {\n            // Control reaches here only when closeFuture is created\n            // and no more connected consumers left.\n            closeFuture.complete(null);\n        }","sourceCodeStart":247,"sourceCodeEnd":283,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java#L247-L283","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["On the client, avoid double-closing consumers (guard close() with an idempotent flag)","Verify the client is connected to the broker that owns the consumer (check lookup/connection logs)","Retry subscribe/consumer creation after this error — the dispatcher state has self-corrected (consumer absent)","If seen frequently with failover, check broker version for known dispatcher race fixes and upgrade"],"exampleFix":"// before\nconsumer.close();\nconsumer.close(); // second close triggers server-side error\n// after\nif (consumer != null && !closed) {\n    consumer.close();\n    closed = true;\n}","handlingStrategy":"try-catch","validationCode":"// client: ensure the consumer is open and connected before close\nif (consumer == null || !consumer.isConnected()) return;","typeGuard":"boolean safeToClose(Consumer<?> c) {\n    return c != null && c.isConnected() && !c.getLastDisconnectedTimestamp().isPresent() == false;\n}","tryCatchPattern":"// server-side throw is ServerMetadataException; clients see it as broker-side error on close\ntry {\n    consumer.close();\n} catch (PulsarClientException e) {\n    log.warn(\"close failed; consumer may have already been removed\", e); // idempotent close\n}","preventionTips":["Make close() idempotent on the client side","Handle reconnects so close is issued to the broker that owns the consumer","Upgrade brokers if dispatcher removal races are observed during failover"],"tags":["pulsar","broker","consumer","state-mismatch","race-condition"],"backgroundTag":"consumer-not-connected","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"}