{"record":{"id":"d69b02746ac85bff","repo":"apache/pulsar","slug":"getting-consumer-is-not-supported","errorCode":null,"errorMessage":"Getting consumer is not supported","messagePattern":"Getting consumer is not supported","errorType":"exception","errorClass":"PulsarClientException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java","lineNumber":759,"sourceCode":"\n    // returns null if consumer not found\n    private Consumer<?> tryGetConsumer(String topic, int partition) {\n        if (partition == 0) {\n            // maybe a non-partitioned topic\n            Consumer<?> consumer = topicConsumers.get(TopicName.get(topic));\n\n            if (consumer != null) {\n                return consumer;\n            }\n        }\n        // maybe partitioned topic\n        return topicConsumers.get(TopicName.get(topic).getPartition(partition));\n    }\n\n    @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}","sourceCodeStart":741,"sourceCodeEnd":777,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java#L741-L777","documentation":"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.","triggerScenarios":"User code calls context.getConsumer(topic, partition) on a function instance where exposeConsumerEnabled is false, so inputConsumers was never initialized.","commonSituations":"Functions that read raw consumers for manual seek/ack deployed without enabling consumer exposure; tests invoking getConsumer on a ContextImpl built without consumers.","solutions":["Enable consumer exposure in the function worker config (`exposeConsumerEnabled=true`) and redeploy.","Use context.getOutputTopic()/newOutputMessage() or a dedicated PulsarConsumerBuilder-style approach instead of accessing input consumers.","Create your own consumer via the PulsarClient obtained from the worker if admin/client exposure is enabled."],"exampleFix":"// before (worker config)\nexposeConsumerEnabled=false\n// after\nexposeConsumerEnabled=true","handlingStrategy":"validation","validationCode":"if (!Boolean.getBoolean(\"pulsar.functions.exposeConsumerEnabled\")) {\n    throw new IllegalStateException(\"getConsumer requires exposeConsumerEnabled=true\");\n}","typeGuard":null,"tryCatchPattern":"try { Consumer<?> c = context.getConsumer(topic, partition); ... } catch (PulsarClientException e) { /* use default record ack path */ }","preventionTips":["Enable exposeConsumerEnabled=true in worker config before using getConsumer","Do not rely on raw consumers in ordinary functions","Wrap consumer access behind a feature-checked helper"],"tags":["pulsar-functions","consumer","configuration","pulsar-client"],"backgroundTag":"consumer-not-exposed","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"}