apache/pulsar · error · PulsarClientException.InvalidConfigurationException

Cannot use receive() when a listener has been set

Error message

Cannot use receive() when a listener has been set

What it means

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.

Source

Thrown at pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java:278

            unAckedMessageTracker.add(messageId, redeliveryCount);
        }
    }

    protected void reduceCurrentReceiverQueueSize() {
        if (!conf.isAutoScaledReceiverQueueSizeEnabled()) {
            return;
        }
        int oldSize = getCurrentReceiverQueueSize();
        int newSize = Math.max(minReceiverQueueSize(), oldSize / 2);
        if (oldSize > newSize) {
            setCurrentReceiverQueueSize(newSize);
        }
    }

    @Override
    public Message<T> receive() throws PulsarClientException {
        if (listener != null) {
            throw new PulsarClientException.InvalidConfigurationException(
                    "Cannot use receive() when a listener has been set");
        }
        verifyConsumerState();
        return internalReceive();
    }

    @Override
    public CompletableFuture<Message<T>> receiveAsync() {
        if (listener != null) {
            return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(
                    "Cannot use receive() when a listener has been set"));
        }
        try {
            verifyConsumerState();
        } catch (PulsarClientException e) {
            return FutureUtil.failedFuture(e);
        }
        return internalReceiveAsync();

View on GitHub (pinned to 820761864e)

Solutions

1. Remove the messageListener from the ConsumerBuilder and consume via receive()/receiveAsync, or
2. Remove the receive() calls and handle messages inside the MessageListener.received() callback
3. If both modes are needed, build two separate consumers (one with a listener, one polling) on the subscription

Example fix

// before
Consumer<String> c = client.newConsumer(Schema.STRING)
    .topic("t").subscriptionName("s")
    .messageListener((consumer, msg) -> handle(msg))
    .subscribe();
Message<String> m = c.receive();
// after
Consumer<String> c = client.newConsumer(Schema.STRING)
    .topic("t").subscriptionName("s")
    .subscribe(); // no listener
Message<String> m = c.receive();

if (consumer.getConsumerConfigurationData().getListener() != null) {
    throw new IllegalStateException("Use listener mode or receive(), not both");
}

boolean canPoll(org.apache.pulsar.client.api.Consumer<T> c) {
    return ((ConsumerBase<T>) c).getConsumerConfigurationData().getListener() == null;
}

try {
    Message<T> msg = consumer.receive();
} catch (PulsarClientException.InvalidConfigurationException e) {
    // consumer is in listener mode; route through listener logic instead
}

When it happens

Trigger: Calling consumer.receive() (or receive(timeout, unit) / receiveAsync) on a consumer that was built with ConsumerBuilder.messageListener(...) set to a non-null listener.

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

Related errors

• Can't use receive with timeout, if the queue size is 0
• Can't use batch receive, if the queue size is 0
• At least one of maxNumMessages, maxNumBytes, timeout must be
• Must set timeout unit for timeout.
• Broker doesn't allow forced deletion of namespaces

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