apache/pulsar · error · RestException

Function in trigger function is not ready

Error message

Function in trigger function is not ready

What it means

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.

Source

Thrown at pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java:1168

        } catch (IllegalArgumentException e) {
            topicFound = false;
        }
        if (functionMetaData.getFunctionDetails().getSource().getInputSpecsCount() == 0
                || !topicFound) {
            log.error().attr("tenant", tenant).attr("namespace", namespace).attr("componentName", functionName)

                    .attr("topic", inputTopicToWrite)

                    .log("Function in trigger function has unidentified topic @ / / /");
            throw new RestException(Status.BAD_REQUEST, "Function in trigger function has unidentified topic");
        }
        try {
            worker().getBrokerAdmin().topics().getSubscriptions(inputTopicToWrite);
        } catch (PulsarAdminException e) {
            log.error().attr("tenant", tenant).attr("namespace", namespace).attr("componentName", functionName)

                    .exception(e).log("Function in trigger function is not ready @ / / /");
            throw new RestException(Status.BAD_REQUEST, "Function in trigger function is not ready");
        }
        String outputTopic = functionMetaData.getFunctionDetails().getSink().getTopic();
        Reader<byte[]> reader = null;
        Producer<byte[]> producer = null;
        try {
            if (!isEmpty(outputTopic)) {
                reader = worker().getClient().newReader()
                        .topic(outputTopic)
                        .startMessageId(MessageId.latest)
                        .readerName(worker().getWorkerConfig().getWorkerId() + "-trigger-"
                                + FunctionCommon.getFullyQualifiedName(tenant, namespace, functionName))
                        .create();
            }
            producer = worker().getClient().newProducer(Schema.AUTO_PRODUCE_BYTES())
                    .topic(inputTopicToWrite)
                    .producerName(worker().getWorkerConfig().getWorkerId() + "-trigger-"
                            + FunctionCommon.getFullyQualifiedName(tenant, namespace, functionName))
                    .create();

View on GitHub (pinned to 820761864e)

Solutions

1. Verify the input topic exists (pulsar-admin topics list) and create it if missing, then retry the trigger.
2. For partitioned topics, wait until all partitions are provisioned before triggering.
3. Check broker health/reachability from the worker (service URL configuration).
4. Ensure the worker's admin client has permissions to look up topics (admin role).

Example fix

// before: trigger against not-yet-created topic
// after: pre-create and verify, then trigger
admin.topics().createPartitionedTopic("persistent://public/default/src", 3);
if (admin.topics().getSubscriptions("persistent://public/default/src") != null) {
    triggerFunction(...);
}

static boolean inputTopicReady(PulsarAdmin admin, String topic) {
  try { admin.topics().getSubscriptions(topic); return true; }
  catch (PulsarAdminException e) { return false; }
}

awaitAtMost(30, SECONDS).until(() -> inputTopicReady(admin, inputTopic));
try { triggerFunction(...); }
catch (PulsarAdminException e) {
  if (e.getResponseStatus() == 400 && e.getMessage().contains("not ready")) retryAfterDelay();
  else throw e;
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Failed to add schema to an active topic with empty(BYTES) sc
• The topic has a max partition index of %d, the number of par
• entryFilterNames can't be empty. To remove entry filters use
• The offloadPolicies must be specified for namespace offload.
• The driver is not supported, support value: ${supportedDrive

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