{"record":{"id":"0ba6a2a313a8d9ec","repo":"apache/pulsar","slug":"function-in-trigger-function-is-not-ready","errorCode":null,"errorMessage":"Function in trigger function is not ready","messagePattern":"Function in trigger function is not ready","errorType":"http","errorClass":"RestException","httpStatus":400,"severity":"error","filePath":"pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java","lineNumber":1168,"sourceCode":"        } catch (IllegalArgumentException e) {\n            topicFound = false;\n        }\n        if (functionMetaData.getFunctionDetails().getSource().getInputSpecsCount() == 0\n                || !topicFound) {\n            log.error().attr(\"tenant\", tenant).attr(\"namespace\", namespace).attr(\"componentName\", functionName)\n\n                    .attr(\"topic\", inputTopicToWrite)\n\n                    .log(\"Function in trigger function has unidentified topic @ / / /\");\n            throw new RestException(Status.BAD_REQUEST, \"Function in trigger function has unidentified topic\");\n        }\n        try {\n            worker().getBrokerAdmin().topics().getSubscriptions(inputTopicToWrite);\n        } catch (PulsarAdminException e) {\n            log.error().attr(\"tenant\", tenant).attr(\"namespace\", namespace).attr(\"componentName\", functionName)\n\n                    .exception(e).log(\"Function in trigger function is not ready @ / / /\");\n            throw new RestException(Status.BAD_REQUEST, \"Function in trigger function is not ready\");\n        }\n        String outputTopic = functionMetaData.getFunctionDetails().getSink().getTopic();\n        Reader<byte[]> reader = null;\n        Producer<byte[]> producer = null;\n        try {\n            if (!isEmpty(outputTopic)) {\n                reader = worker().getClient().newReader()\n                        .topic(outputTopic)\n                        .startMessageId(MessageId.latest)\n                        .readerName(worker().getWorkerConfig().getWorkerId() + \"-trigger-\"\n                                + FunctionCommon.getFullyQualifiedName(tenant, namespace, functionName))\n                        .create();\n            }\n            producer = worker().getClient().newProducer(Schema.AUTO_PRODUCE_BYTES())\n                    .topic(inputTopicToWrite)\n                    .producerName(worker().getWorkerConfig().getWorkerId() + \"-trigger-\"\n                            + FunctionCommon.getFullyQualifiedName(tenant, namespace, functionName))\n                    .create();","sourceCodeStart":1150,"sourceCodeEnd":1186,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java#L1150-L1186","documentation":"triggerFunction calls brokerAdmin.topics().getSubscriptions(inputTopic) to confirm the input topic is usable before injecting a message. A PulsarAdminException here (topic not found, broker unreachable, no partitions yet) is converted to HTTP 400 'Function in trigger function is not ready' — the function exists but its input topic isn't serviceable yet.","triggerScenarios":"Input topic does not exist or was deleted; broker hosting the topic is down/unreachable; topic still being created (partitioned topic not yet provisioned); admin credentials insufficient to look up the topic.","commonSituations":"Triggering immediately after creating a partitioned topic before partitions materialize; dev environment where the source topic was never created; misconfigured broker service URL on the worker.","solutions":["Verify the input topic exists (pulsar-admin topics list) and create it if missing, then retry the trigger.","For partitioned topics, wait until all partitions are provisioned before triggering.","Check broker health/reachability from the worker (service URL configuration).","Ensure the worker's admin client has permissions to look up topics (admin role)."],"exampleFix":"// before: trigger against not-yet-created topic\n// after: pre-create and verify, then trigger\nadmin.topics().createPartitionedTopic(\"persistent://public/default/src\", 3);\nif (admin.topics().getSubscriptions(\"persistent://public/default/src\") != null) {\n    triggerFunction(...);\n}","handlingStrategy":"retry","validationCode":"static boolean inputTopicReady(PulsarAdmin admin, String topic) {\n  try { admin.topics().getSubscriptions(topic); return true; }\n  catch (PulsarAdminException e) { return false; }\n}","typeGuard":null,"tryCatchPattern":"awaitAtMost(30, SECONDS).until(() -> inputTopicReady(admin, inputTopic));\ntry { triggerFunction(...); }\ncatch (PulsarAdminException e) {\n  if (e.getResponseStatus() == 400 && e.getMessage().contains(\"not ready\")) retryAfterDelay();\n  else throw e;\n}","preventionTips":["Pre-create input topics (including all partitions) before triggering","Wait for topic lookup to succeed in CI before the trigger step","Monitor broker availability from the worker network path"],"tags":["pulsar","broker","topic","not-ready"],"backgroundTag":"topic-not-found","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}