apache/pulsar · error · NotImplementedException

AutoScaledReceiverQueueSize is not supported in ZeroQueueCon

Error message

AutoScaledReceiverQueueSize is not supported in ZeroQueueConsumerImpl

What it means

ZeroQueueConsumerImpl.initReceiverQueueSize() throws NotImplementedException when auto-scaled receiver queue size is enabled on a consumer configured with receiverQueueSize=0. Zero-queue consumers have a fixed queue of 0 (fetch-per-message mode); the auto-scaled queue feature fundamentally requires a non-zero queue to scale, so combining them is rejected at consumer construction.

Source

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

    public ZeroQueueConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurationData<T> conf,
             ExecutorProvider executorProvider, int partitionIndex, boolean hasParentConsumer,
             CompletableFuture<Consumer<T>> subscribeFuture, MessageId startMessageId, Schema<T> schema,
             ConsumerInterceptors<T> interceptors,
             boolean createTopicIfDoesNotExist) {
        super(client, topic, conf, executorProvider, partitionIndex, hasParentConsumer, false, subscribeFuture,
                startMessageId, 0 /* startMessageRollbackDurationInSec */, schema, interceptors,
                createTopicIfDoesNotExist);
    }

    @Override
    public int minReceiverQueueSize() {
        return 0;
    }

    @Override
    public void initReceiverQueueSize() {
        if (conf.isAutoScaledReceiverQueueSizeEnabled()) {
            throw new NotImplementedException("AutoScaledReceiverQueueSize is not supported in ZeroQueueConsumerImpl");
        } else {
            CURRENT_RECEIVER_QUEUE_SIZE_UPDATER.set(this, 0);
        }
    }

    @Override
    protected Message<T> internalReceive() throws PulsarClientException {
        zeroQueueLock.lock();
        try {
            Message<T> msg = fetchSingleMessageFromBroker();
            trackMessage(msg);
            return beforeConsume(msg);
        } finally {
            zeroQueueLock.unlock();
        }
    }

    @Override

View on GitHub (pinned to 820761864e)

Solutions

Disable autoScaledReceiverQueueSizeEnabled in the configuration when using zero-queue consumers.
Remove receiverQueueSize(0) (use a positive size) if auto-scaling is desired.
Guard construction: only enable auto-scaled queue when conf.getReceiverQueueSize() > 0.

Example fix

// before
client.newConsumer().receiverQueueSize(0) // conf.autoScaledReceiverQueueSizeEnabled=true -> throws
// after
client.newConsumer().receiverQueueSize(0)
      .loadConf(Map.of("autoScaledReceiverQueueSizeEnabled", false));

Defensive patterns

Strategy: validation

Validate before calling

if (conf.getReceiverQueueSize() == 0 && conf.isAutoScaledReceiverQueueSizeEnabled()) {
    throw new IllegalArgumentException("autoScaledReceiverQueueSize cannot be used with receiverQueueSize=0");
}

Try / catch

try {
    Consumer<T> c = client.newConsumer()...subscribe();
} catch (NotImplementedException e) {
    log.error("incompatible consumer options: {}", e.getMessage());
    throw e;
}

Prevention

Never combine receiverQueueSize=0 with autoScaledReceiverQueueSizeEnabled=true.
Gate the auto-scaled flag on a positive receiver queue size in shared config code.
Document that zero-queue consumers are fetch-per-message and cannot auto-scale.

When it happens

Trigger: Creating a consumer/reader with ConsumerBuilder.receiverQueueSize(0) (or conf receiverQueueSize=0) while autoScaledReceiverQueueSizeEnabled=true in the configuration data; the factory's initReceiverQueueSize() runs at construction and throws immediately.

Common situations: A shared ClientConfigurationData enables auto-scaled receiver queue globally while some consumers/readers opt into zero queue size; readers with receiverQueueSize=0 for at-least-once semantics; configuration merges turning on the feature accidentally; MultiTopicsConsumer passing zero per-topic queue sizes.

Related errors

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