apache/pulsar · error · PulsarClientException

Getting consumer is not supported

Error message

Getting consumer is not supported

What it means

getConsumer(topic, partition) requires the instance's inputConsumers map, which is only populated when the instance was set up to expose consumers (exposeConsumerEnabled). When inputConsumers is null, the instance does not own consumers for user code, and the method throws PulsarClientException("Getting consumer is not supported") instead of returning a consumer.

Source

Thrown at pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java:759

    // returns null if consumer not found
    private Consumer<?> tryGetConsumer(String topic, int partition) {
        if (partition == 0) {
            // maybe a non-partitioned topic
            Consumer<?> consumer = topicConsumers.get(TopicName.get(topic));

            if (consumer != null) {
                return consumer;
            }
        }
        // maybe partitioned topic
        return topicConsumers.get(TopicName.get(topic).getPartition(partition));
    }

    @VisibleForTesting
    Consumer<?> getConsumer(String topic, int partition) throws PulsarClientException {
        if (inputConsumers == null) {
            throw new PulsarClientException("Getting consumer is not supported");
        }

        Consumer<?> consumer = tryGetConsumer(topic, partition);
        if (consumer == null) {
            // MultiTopicsConsumer's list of consumers could change
            // if partitions changed or pattern(s) used to subscribe.
            // Reload and try one more time.
            reloadConsumersFromMultiTopicsConsumers();
            consumer = tryGetConsumer(topic, partition);
        }

        if (consumer != null) {
            return consumer;
        }
        throw new PulsarClientException("Consumer for topic " + topic
                + " partition " + partition + " is not found");
    }
}

View on GitHub (pinned to 820761864e)

Solutions

1. Enable consumer exposure in the function worker config (`exposeConsumerEnabled=true`) and redeploy.
2. Use context.getOutputTopic()/newOutputMessage() or a dedicated PulsarConsumerBuilder-style approach instead of accessing input consumers.
3. Create your own consumer via the PulsarClient obtained from the worker if admin/client exposure is enabled.

Example fix

// before (worker config)
exposeConsumerEnabled=false
// after
exposeConsumerEnabled=true

if (!Boolean.getBoolean("pulsar.functions.exposeConsumerEnabled")) {
    throw new IllegalStateException("getConsumer requires exposeConsumerEnabled=true");
}

try { Consumer<?> c = context.getConsumer(topic, partition); ... } catch (PulsarClientException e) { /* use default record ack path */ }

When it happens

Trigger: User code calls context.getConsumer(topic, partition) on a function instance where exposeConsumerEnabled is false, so inputConsumers was never initialized.

Common situations: Functions that read raw consumers for manual seek/ack deployed without enabling consumer exposure; tests invoking getConsumer on a ContextImpl built without consumers.

Related errors

• AutoScaledReceiverQueueSize is not supported in ZeroQueueCon
• Consumer for topic %s partition %s is not found
• pulsarServiceUrl cannot be null
• Cannot use receive() when a listener has been set
• Can't use receive with timeout, if the queue size is 0

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