{"record":{"id":"0873ba07113edce4","repo":"apache/pulsar","slug":"receiver-queue-size-can-t-be-changed-in-zeroqueuec","errorCode":null,"errorMessage":"Receiver queue size can't be changed in ZeroQueueConsumerImpl","messagePattern":"Receiver queue size can't be changed in ZeroQueueConsumerImpl","errorType":"exception","errorClass":"NotImplementedException","httpStatus":null,"severity":"error","filePath":"pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java","lineNumber":212,"sourceCode":"    protected void tryTriggerListener() {\n        // Ignore since it was already triggered in the triggerZeroQueueSizeListener() call\n    }\n\n    @Override\n    void receiveIndividualMessagesFromBatch(BrokerEntryMetadata brokerEntryMetadata, MessageMetadata msgMetadata,\n                                            int redeliveryCount, long[] ackSet, ByteBuf uncompressedPayload,\n                                            MessageIdData messageId, ClientCnx cnx, long consumerEpoch,\n                                            boolean isEncrypted) {\n\n        rejectBatchMessageByClosingConsumer(\n                new MessageIdImpl(messageId.getLedgerId(), messageId.getEntryId(), getPartitionIndex())\n        );\n    }\n\n    @Override\n    protected void setCurrentReceiverQueueSize(int newSize) {\n        //receiver queue size is fixed as 0.\n        throw new NotImplementedException(\"Receiver queue size can't be changed in ZeroQueueConsumerImpl\");\n    }\n\n    @Override\n    protected void processPayloadByProcessor(BrokerEntryMetadata brokerEntryMetadata,\n                                             MessageMetadata messageMetadata, ByteBuf byteBuf,\n                                             MessageIdImpl messageId, Schema<T> schema,\n                                             int redeliveryCount, List<Long> ackSet, long consumerEpoch) {\n        if (this.isBatch(messageMetadata)) {\n            rejectBatchMessageByClosingConsumer(messageId);\n        } else {\n            super.processPayloadByProcessor(brokerEntryMetadata, messageMetadata, byteBuf, messageId, schema,\n                    redeliveryCount, ackSet, consumerEpoch);\n        }\n    }\n\n    private void rejectBatchMessageByClosingConsumer(MessageIdImpl messageId) {\n        log.warn().attr(\"messageId\", messageId)\n                .log(\"Closing consumer - due to unsupported received batch-message with zero receiver queue size\");","sourceCodeStart":194,"sourceCodeEnd":230,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java#L194-L230","documentation":"ZeroQueueConsumerImpl overrides setCurrentReceiverQueueSize(int) to always throw NotImplementedException, because the receiver queue of a zero-queue consumer is fixed at 0 by design and cannot be resized at runtime.","triggerScenarios":"Calling consumer.setCurrentReceiverQueueSize(n) on a consumer created with receiverQueueSize=0; also hit by internal paths (e.g. multi-topic consumer adjustments or reconnect logic) that attempt to resize the queue on a zero-queue consumer.","commonSituations":"Applications that dynamically tune receiver queue size under backpressure but also run some zero-queue consumers; generic consumer-management code that calls setCurrentReceiverQueueSize on all consumers without checking their type/config.","solutions":["Skip the call when the consumer is a zero-queue consumer (check getReceiverQueueSize() == 0 before calling).","Recreate the consumer with the desired receiverQueueSize instead of resizing at runtime.","Use a positive receiverQueueSize at creation if runtime resizing is required."],"exampleFix":"// before\nconsumer.setCurrentReceiverQueueSize(newSize); // throws for zero-queue consumers\n// after\nif (consumer.getReceiverQueueSize() > 0) {\n    consumer.setCurrentReceiverQueueSize(newSize);\n}","handlingStrategy":"type-guard","validationCode":"if (consumer.getReceiverQueueSize() > 0) {\n    consumer.setCurrentReceiverQueueSize(newSize);\n}","typeGuard":"boolean isResizable(Consumer<?> c) { return !(c instanceof ZeroQueueConsumerImpl) && c.getReceiverQueueSize() > 0; }","tryCatchPattern":"try {\n    consumer.setCurrentReceiverQueueSize(newSize);\n} catch (NotImplementedException e) {\n    log.warn(\"receiver queue is fixed for this consumer; recreate it to change size\");\n}","preventionTips":["Check getReceiverQueueSize() before any runtime resize call.","Recreate the consumer with the desired queue size instead of resizing zero-queue consumers.","Centralize queue-size mutation behind a helper that skips zero-queue consumers."],"tags":["java","consumer","unsupported-operation","pulsar-client"],"backgroundTag":"unsupported-operation","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"}