apache/pulsar · error · IllegalArgumentException

Input topic %s is invalid

Error message

Input topic %s is invalid

What it means

Each input topic collected from the sink config is validated with TopicName.isValid(); a topic that isn't a syntactically valid Pulsar topic name (missing domain or malformed tenant/namespace/local-name) causes this IllegalArgumentException with the offending topic name interpolated.

Source

Thrown at pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java:436

        if (isEmpty(sinkConfig.getTenant())) {
            throw new IllegalArgumentException("Sink tenant cannot be null");
        }
        if (isEmpty(sinkConfig.getNamespace())) {
            throw new IllegalArgumentException("Sink namespace cannot be null");
        }
        if (isEmpty(sinkConfig.getName())) {
            throw new IllegalArgumentException("Sink name cannot be null");
        }

        // make we sure we have one source of input
        Collection<String> allInputs = collectAllInputTopics(sinkConfig);
        if (allInputs.isEmpty()) {
            throw new IllegalArgumentException("Must specify at least one topic of input via topicToSerdeClassName, "
                    + "topicsPattern, topicToSchemaType or inputSpecs");
        }
        for (String topic : allInputs) {
            if (!TopicName.isValid(topic)) {
                throw new IllegalArgumentException(String.format("Input topic %s is invalid", topic));
            }
        }
        if (!isEmpty(sinkConfig.getLogTopic())) {
            if (!TopicName.isValid(sinkConfig.getLogTopic())) {
                throw new IllegalArgumentException(
                        String.format("LogTopic topic %s is invalid", sinkConfig.getLogTopic()));
            }
        }

        if (sinkConfig.getParallelism() != null && sinkConfig.getParallelism() <= 0) {
            throw new IllegalArgumentException("Sink parallelism must be a positive number");
        }

        if (sinkConfig.getResources() != null) {
            ResourceConfigUtils.validate(sinkConfig.getResources());
        }

        if (sinkConfig.getTimeoutMs() != null && sinkConfig.getTimeoutMs() < 0) {

View on GitHub (pinned to 820761864e)

Solutions

1. Use a fully-qualified topic name: persistent://<tenant>/<namespace>/<topic>.
2. Trim whitespace and re-check for typos/slashes in the config value.
3. Validate locally with TopicName.isValid(topic) before submitting.
4. If using a pattern input, verify the pattern is a valid topic-pattern string.

Example fix

// before
inputSpecs.put("my-topic", new ConsumerConfig());
// after
inputSpecs.put("persistent://public/default/my-topic", new ConsumerConfig());

for (String topic : collectInputs(sinkConfig)) {
    if (!org.apache.pulsar.common.naming.TopicName.isValid(topic)) {
        throw new IllegalArgumentException("invalid input topic: " + topic);
    }
}

static boolean isValidTopic(String topic) {
    return topic != null && org.apache.pulsar.common.naming.TopicName.isValid(topic.trim());
}

try {
    SinkConfigUtils.validateAndExtractDetails(cfg, sinkPkg, transformPkg, true);
} catch (IllegalArgumentException e) {
    if (e.getMessage().startsWith("Input topic")) {
        log.error("Malformed input topic in config: {}", e.getMessage(), e);
    }
    throw e;
}

When it happens

Trigger: Input topic strings in topicToSerdeClassName, inputSpecs, topicsToSerdeClassName, or (short-form) topicName that lack the persistent:// or topic:// prefix or contain invalid characters/segments, e.g. 'my-topic' instead of 'persistent://public/default/my-topic' — though short names are typically completed first, malformed ones fail.

Common situations: Typing a bare topic name where a fully-qualified name is required; extra slashes or spaces in the topic string; YAML parsing turning a topic into a non-string; using 'persistent:/default/topic' (one slash).

Related errors

• LogTopic topic %s is invalid
• Invalid short topic name '${completeTopicName}', it should b
• V1 topic names (with cluster component) are no longer suppor
• Invalid topic name ${completeTopicName}
• Invalid segment topic name: local name must contain '<parent

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