{"record":{"id":"6142807f6035e648","repo":"apache/pulsar","slug":"cannot-use-receive-when-a-listener-has-been-set","errorCode":null,"errorMessage":"Cannot use receive() when a listener has been set","messagePattern":"Cannot use receive\\(\\) when a listener has been set","errorType":"exception","errorClass":"PulsarClientException.InvalidConfigurationException","httpStatus":null,"severity":"error","filePath":"pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java","lineNumber":278,"sourceCode":"            unAckedMessageTracker.add(messageId, redeliveryCount);\n        }\n    }\n\n    protected void reduceCurrentReceiverQueueSize() {\n        if (!conf.isAutoScaledReceiverQueueSizeEnabled()) {\n            return;\n        }\n        int oldSize = getCurrentReceiverQueueSize();\n        int newSize = Math.max(minReceiverQueueSize(), oldSize / 2);\n        if (oldSize > newSize) {\n            setCurrentReceiverQueueSize(newSize);\n        }\n    }\n\n    @Override\n    public Message<T> receive() throws PulsarClientException {\n        if (listener != null) {\n            throw new PulsarClientException.InvalidConfigurationException(\n                    \"Cannot use receive() when a listener has been set\");\n        }\n        verifyConsumerState();\n        return internalReceive();\n    }\n\n    @Override\n    public CompletableFuture<Message<T>> receiveAsync() {\n        if (listener != null) {\n            return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(\n                    \"Cannot use receive() when a listener has been set\"));\n        }\n        try {\n            verifyConsumerState();\n        } catch (PulsarClientException e) {\n            return FutureUtil.failedFuture(e);\n        }\n        return internalReceiveAsync();","sourceCodeStart":260,"sourceCodeEnd":296,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java#L260-L296","documentation":"Pulsar consumers support two mutually exclusive delivery models: listener-based (push, via messageListener on ConsumerBuilder) and polling-based (pull, via receive()). ConsumerBase.receive() throws InvalidConfigurationException when a listener is set because the listener already consumes messages from the internal queue, leaving nothing (or competing messages) for receive() calls. This is a configuration conflict detected at runtime, not a transient failure.","triggerScenarios":"Calling consumer.receive() (or receive(timeout, unit) / receiveAsync) on a consumer that was built with ConsumerBuilder.messageListener(...) set to a non-null listener.","commonSituations":"Copy-pasting sample polling code into an app whose consumer factory already registers a MessageListener; adding a listener for metrics/monitoring to an existing polling consumer; shared consumer-creation helper that optionally attaches a listener.","solutions":["Remove the messageListener from the ConsumerBuilder and consume via receive()/receiveAsync, or","Remove the receive() calls and handle messages inside the MessageListener.received() callback","If both modes are needed, build two separate consumers (one with a listener, one polling) on the subscription"],"exampleFix":"// before\nConsumer<String> c = client.newConsumer(Schema.STRING)\n    .topic(\"t\").subscriptionName(\"s\")\n    .messageListener((consumer, msg) -> handle(msg))\n    .subscribe();\nMessage<String> m = c.receive();\n// after\nConsumer<String> c = client.newConsumer(Schema.STRING)\n    .topic(\"t\").subscriptionName(\"s\")\n    .subscribe(); // no listener\nMessage<String> m = c.receive();","handlingStrategy":"validation","validationCode":"if (consumer.getConsumerConfigurationData().getListener() != null) {\n    throw new IllegalStateException(\"Use listener mode or receive(), not both\");\n}","typeGuard":"boolean canPoll(org.apache.pulsar.client.api.Consumer<T> c) {\n    return ((ConsumerBase<T>) c).getConsumerConfigurationData().getListener() == null;\n}","tryCatchPattern":"try {\n    Message<T> msg = consumer.receive();\n} catch (PulsarClientException.InvalidConfigurationException e) {\n    // consumer is in listener mode; route through listener logic instead\n}","preventionTips":["Decide push vs pull model per subscription at builder time and enforce it in your consumer factory","Never set messageListener in shared consumer-creation helpers used by polling code","Wrap consumer creation in a factory that takes a mode enum (LISTENER|POLLING) and configures exclusively"],"tags":["pulsar","configuration","consumer","invalid-configuration"],"backgroundTag":"listener-and-polling-conflict","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"}