apache/pulsar · error · NotImplementedException

Receiver queue size can't be changed in ZeroQueueConsumerImp

Error message

Receiver queue size can't be changed in ZeroQueueConsumerImpl

What it means

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.

Source

Thrown at pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java:212

    protected void tryTriggerListener() {
        // Ignore since it was already triggered in the triggerZeroQueueSizeListener() call
    }

    @Override
    void receiveIndividualMessagesFromBatch(BrokerEntryMetadata brokerEntryMetadata, MessageMetadata msgMetadata,
                                            int redeliveryCount, long[] ackSet, ByteBuf uncompressedPayload,
                                            MessageIdData messageId, ClientCnx cnx, long consumerEpoch,
                                            boolean isEncrypted) {

        rejectBatchMessageByClosingConsumer(
                new MessageIdImpl(messageId.getLedgerId(), messageId.getEntryId(), getPartitionIndex())
        );
    }

    @Override
    protected void setCurrentReceiverQueueSize(int newSize) {
        //receiver queue size is fixed as 0.
        throw new NotImplementedException("Receiver queue size can't be changed in ZeroQueueConsumerImpl");
    }

    @Override
    protected void processPayloadByProcessor(BrokerEntryMetadata brokerEntryMetadata,
                                             MessageMetadata messageMetadata, ByteBuf byteBuf,
                                             MessageIdImpl messageId, Schema<T> schema,
                                             int redeliveryCount, List<Long> ackSet, long consumerEpoch) {
        if (this.isBatch(messageMetadata)) {
            rejectBatchMessageByClosingConsumer(messageId);
        } else {
            super.processPayloadByProcessor(brokerEntryMetadata, messageMetadata, byteBuf, messageId, schema,
                    redeliveryCount, ackSet, consumerEpoch);
        }
    }

    private void rejectBatchMessageByClosingConsumer(MessageIdImpl messageId) {
        log.warn().attr("messageId", messageId)
                .log("Closing consumer - due to unsupported received batch-message with zero receiver queue size");

View on GitHub (pinned to 820761864e)

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.

Example fix

// before
consumer.setCurrentReceiverQueueSize(newSize); // throws for zero-queue consumers
// after
if (consumer.getReceiverQueueSize() > 0) {
    consumer.setCurrentReceiverQueueSize(newSize);
}

Defensive patterns

Strategy: type-guard

Validate before calling

if (consumer.getReceiverQueueSize() > 0) {
    consumer.setCurrentReceiverQueueSize(newSize);
}

Type guard

boolean isResizable(Consumer<?> c) { return !(c instanceof ZeroQueueConsumerImpl) && c.getReceiverQueueSize() > 0; }

Try / catch

try {
    consumer.setCurrentReceiverQueueSize(newSize);
} catch (NotImplementedException e) {
    log.warn("receiver queue is fixed for this consumer; recreate it to change size");
}

Prevention

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.

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

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