{"record":{"id":"f7667b7e6f14e92d","repo":"apache/pulsar","slug":"can-t-use-receive-with-timeout-if-the-queue-size","errorCode":null,"errorMessage":"Can't use receive with timeout, if the queue size is 0","messagePattern":"Can't use receive with timeout, if the queue size is 0","errorType":"exception","errorClass":"PulsarClientException.InvalidConfigurationException","httpStatus":null,"severity":"error","filePath":"pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java","lineNumber":306,"sourceCode":"            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();\n    }\n\n    protected abstract Message<T> internalReceive() throws PulsarClientException;\n\n    protected abstract CompletableFuture<Message<T>> internalReceiveAsync();\n\n    @Override\n    public Message<T> receive(int timeout, TimeUnit unit) throws PulsarClientException {\n        if (getCurrentReceiverQueueSize() == 0) {\n            throw new PulsarClientException.InvalidConfigurationException(\n                    \"Can't use receive with timeout, if the queue size is 0\");\n        }\n        if (listener != null) {\n            throw new PulsarClientException.InvalidConfigurationException(\n                    \"Cannot use receive() when a listener has been set\");\n        }\n\n        verifyConsumerState();\n        return internalReceive(timeout, unit);\n    }\n\n    protected abstract Message<T> internalReceive(long timeout, TimeUnit unit) throws PulsarClientException;\n\n    @Override\n    public Messages<T> batchReceive() throws PulsarClientException {\n        verifyBatchReceive();\n        verifyConsumerState();\n        return internalBatchReceive();","sourceCodeStart":288,"sourceCodeEnd":324,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java#L288-L324","documentation":"A consumer created with receiverQueueSize=0 (typically for readers or exactly-once / seek-heavy patterns) has no internal buffer to hold a message for the timeout-based receive to poll from, so receive(int timeout, TimeUnit unit) rejects this configuration with InvalidConfigurationException. Queue-size-0 consumers only deliver via async/internal paths, so the timed synchronous receive is unsupported by design.","triggerScenarios":"Building a consumer with ConsumerBuilder.receiverQueueSize(0) (or ConsumerConfigurationData with receiverQueueSize 0, common for Readers) and then calling consumer.receive(timeout, unit).","commonSituations":"Using a Reader with receiverQueueSize=0 for low-latency tailing and then calling the timed receive; tuning queue size to 0 to 'reduce latency' without realizing timed receive stops working; shared consumer builder where queue size 0 is set conditionally.","solutions":["Set receiverQueueSize to at least 1 in the ConsumerBuilder/ReaderBuilder","Use receive() or receiveAsync() instead of the timed receive(timeout, unit)","Use Reader.readNext(timeout, unit), which is designed for reader-style consumption"],"exampleFix":"// before\nReader<String> r = client.newReader(Schema.STRING)\n    .topic(\"t\").readerName(\"rd\")\n    .receiverQueueSize(0).create();\nMessage<String> m = consumer.receive(1, TimeUnit.SECONDS);\n// after\nReader<String> r = client.newReader(Schema.STRING)\n    .topic(\"t\").readerName(\"rd\")\n    .receiverQueueSize(0).create();\nMessage<String> m = r.readNext(1, TimeUnit.SECONDS);","handlingStrategy":"validation","validationCode":"// check before using timed receive\n// receiverQueueSize is fixed at build time; keep a flag from your builder\nif (receiverQueueSize == 0) {\n    throw new IllegalStateException(\"Timed receive() unsupported with receiverQueueSize=0; use Reader.readNext\");\n}","typeGuard":"boolean supportsTimedReceive(org.apache.pulsar.client.api.Consumer<T> c) {\n    return ((ConsumerBase<T>) c).getCurrentReceiverQueueSize() > 0;\n}","tryCatchPattern":"try {\n    Message<T> msg = consumer.receive(1, TimeUnit.SECONDS);\n} catch (PulsarClientException.InvalidConfigurationException e) {\n    // queue size 0: switch to receive() / receiveAsync() / Reader.readNext\n}","preventionTips":["Only set receiverQueueSize(0) on Reader/when you truly need no buffering","Remember Readers must use readNext(), not consumer receive()","Document receiverQueueSize choices in your consumer configuration layer"],"tags":["pulsar","configuration","receiver-queue-size","consumer"],"backgroundTag":"receiver-queue-size-zero","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}