{"record":{"id":"9ee300e9a2c83b69","repo":"apache/pulsar","slug":"consumer-for-topic-s-partition-s-is-not-found","errorCode":null,"errorMessage":"Consumer for topic %s partition %s is not found","messagePattern":"Consumer for topic (.+?) partition (.+?) is not found","errorType":"exception","errorClass":"PulsarClientException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java","lineNumber":774,"sourceCode":"    @VisibleForTesting\n    Consumer<?> getConsumer(String topic, int partition) throws PulsarClientException {\n        if (inputConsumers == null) {\n            throw new PulsarClientException(\"Getting consumer is not supported\");\n        }\n\n        Consumer<?> consumer = tryGetConsumer(topic, partition);\n        if (consumer == null) {\n            // MultiTopicsConsumer's list of consumers could change\n            // if partitions changed or pattern(s) used to subscribe.\n            // Reload and try one more time.\n            reloadConsumersFromMultiTopicsConsumers();\n            consumer = tryGetConsumer(topic, partition);\n        }\n\n        if (consumer != null) {\n            return consumer;\n        }\n        throw new PulsarClientException(\"Consumer for topic \" + topic\n                + \" partition \" + partition + \" is not found\");\n    }\n}\n","sourceCodeStart":756,"sourceCodeEnd":778,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java#L756-L778","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Verify the topic name and partition index match one of the function's configured input topics (and its partition count).","Use tryGetConsumer-like logic or catch the PulsarClientException and retry, since MultiTopicsConsumer children can change dynamically.","Add the desired topic to the function's input subscriptions if it must be consumed there."],"exampleFix":"// before\nConsumer<?> c = context.getConsumer(\"persistent://public/default/my-topic\", 0);\n// after\ntry {\n    Consumer<?> c = context.getConsumer(\"persistent://public/default/my-topic-partition-0\", 0);\n} catch (PulsarClientException e) {\n    // fall back to record.ack()/default processing\n}","handlingStrategy":"retry","validationCode":"boolean isSubscribedInput = functionInputs.stream()\n    .anyMatch(t -> TopicName.get(t).getPartition(idx) == TopicName.get(topic).getPartition(idx));","typeGuard":null,"tryCatchPattern":"try { return context.getConsumer(topic, partition); } catch (PulsarClientException e) { // consumer may appear after rebalance\n  return retryGetConsumer(topic, partition, 3); }","preventionTips":["Only request consumers for topics declared in the function's input spec","Match partition indices against the actual topic partition count","Expect MultiTopicsConsumer child consumers to change; handle misses gracefully"],"tags":["pulsar-functions","consumer","topic","pulsar-client"],"backgroundTag":"consumer-not-found","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}