apache/pulsar · error · PulsarClientException

Consumer for topic %s partition %s is not found

Error message

Consumer for topic %s partition %s is not found

What it means

After confirming consumers are exposed, getConsumer(topic, partition) looks up the per-partition consumer in the inputConsumers map. If no consumer is registered for that exact topic+partition (never subscribed, or a MultiTopicsConsumer whose child consumers changed), it throws PulsarClientException naming the topic and partition.

Source

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

    @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. Verify the topic name and partition index match one of the function's configured input topics (and its partition count).
2. Use tryGetConsumer-like logic or catch the PulsarClientException and retry, since MultiTopicsConsumer children can change dynamically.
3. Add the desired topic to the function's input subscriptions if it must be consumed there.

Example fix

// before
Consumer<?> c = context.getConsumer("persistent://public/default/my-topic", 0);
// after
try {
    Consumer<?> c = context.getConsumer("persistent://public/default/my-topic-partition-0", 0);
} catch (PulsarClientException e) {
    // fall back to record.ack()/default processing
}

boolean isSubscribedInput = functionInputs.stream()
    .anyMatch(t -> TopicName.get(t).getPartition(idx) == TopicName.get(topic).getPartition(idx));

try { return context.getConsumer(topic, partition); } catch (PulsarClientException e) { // consumer may appear after rebalance
  return retryGetConsumer(topic, partition, 3); }

When it happens

Trigger: context.getConsumer("some-topic", n) is called with a topic that is not among the function's subscribed input topics, or with a partition index that doesn't exist / whose underlying consumer moved (partition changes, pattern subscription rebalance).

Common situations: Typos in the topic name; requesting a partition of a non-partitioned topic; accessing topics not listed in the function's input spec; race after topic partition increases.

Related errors

• Getting consumer is not supported
• AutoScaledReceiverQueueSize is not supported in ZeroQueueCon
• Receiver queue size can't be changed in ZeroQueueConsumerImp
• not implemented
• pulsarServiceUrl cannot be null

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