apache/pulsar · error · IllegalArgumentException

LogTopic topic %s is invalid

Error message

LogTopic topic %s is invalid

What it means

If the sink config sets a logTopic, it must be a valid Pulsar topic name. validateAndExtractDetails checks TopicName.isValid on the configured log topic and throws this IllegalArgumentException when it fails, with the bad value interpolated into the message.

Source

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

        }
        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) {
            throw new IllegalArgumentException("Sink timeout must be a positive number");
        }

        String sinkClassName = sinkConfig.getClassName();
        // if class name in sink config is not set, this should be a built-in sink

View on GitHub (pinned to 820761864e)

Solutions

1. Set a fully-qualified log topic: persistent://public/default/sink-logs.
2. Remove the logTopic field if logging to a topic is not needed.
3. Validate with TopicName.isValid before submission.
4. Check YAML quoting — unquoted values with special characters can corrupt the string.

Example fix

// before
sinkConfig.setLogTopic("sink-logs");
// after
sinkConfig.setLogTopic("persistent://public/default/sink-logs");

if (sinkConfig.getLogTopic() != null
    && !org.apache.pulsar.common.naming.TopicName.isValid(sinkConfig.getLogTopic())) {
    throw new IllegalArgumentException("invalid logTopic: " + sinkConfig.getLogTopic());
}

static boolean hasValidLogTopic(SinkConfig cfg) {
    return cfg.getLogTopic() == null
        || org.apache.pulsar.common.naming.TopicName.isValid(cfg.getLogTopic().trim());
}

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

When it happens

Trigger: SinkConfig.setLogTopic("bad name") or a logTopic YAML value that isn't a valid topic (missing domain prefix, invalid characters).

Common situations: Setting logTopic to a bare short name in contexts where it isn't completed; typos in the logTopic field; copying log-topic config between functions and sinks with different expectations.

Related errors

• Input 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/0c9c7a3e883dc21b. Report an issue: GitHub.