apache/pulsar · error · IllegalArgumentException

Must specify at least one topic of input via topicToSerdeCla

Error message

Must specify at least one topic of input via topicToSerdeClassName, topicsPattern, topicToSchemaType or inputSpecs

What it means

A sink must have at least one input topic to consume from. validateAndExtractDetails collects inputs from topicToSerdeClassName, topicsPattern, topicToSchemaType, and inputSpecs (plus the legacy topicsToSerdeClassName/targetTopic), and throws IllegalArgumentException when none are specified.

Source

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

    public static ExtractedSinkDetails validateAndExtractDetails(SinkConfig sinkConfig,
                                                                 ValidatableFunctionPackage sinkFunction,
                                                                 ValidatableFunctionPackage transformFunction,
                                                                 boolean validateConnectorConfig) {
        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");
        }

View on GitHub (pinned to 820761864e)

Solutions

1. Set at least one input: cfg.setTopicPattern("persistent://public/default/my-topic") or cfg.setInputSpecs(Map.of("my-topic", new ConsumerConfig())).
2. Or set topicsToSerdeClassName / topicToSchemaType with the source topic(s).
3. For a single-topic sink, set topicName plus its serde/schema config.
4. List your inputs with the pulsar-admin client or 'pulsar-admin topics list' to confirm the topic names exist and are correctly tenant/namespace-qualified.

Example fix

// before
SinkConfig cfg = new SinkConfig();
cfg.setTenant("public");
cfg.setNamespace("default");
cfg.setName("my-sink");
cfg.setClassName("org.example.MySink");
// after
SinkConfig cfg = new SinkConfig();
cfg.setTenant("public");
cfg.setNamespace("default");
cfg.setName("my-sink");
cfg.setClassName("org.example.MySink");
Map<String, ConsumerConfig> inputSpecs = new HashMap<>();
inputSpecs.put("persistent://public/default/my-topic", new ConsumerConfig());
cfg.setInputSpecs(inputSpecs);

boolean hasInputs = sinkConfig.getInputSpecs() != null && !sinkConfig.getInputSpecs().isEmpty()
    || (sinkConfig.getTopicsToSerdeClassName() != null && !sinkConfig.getTopicsToSerdeClassName().isEmpty())
    || (sinkConfig.getTopicToSchemaType() != null && !sinkConfig.getTopicToSchemaType().isEmpty())
    || sinkConfig.getTopicsPattern() != null;
if (!hasInputs) {
    throw new IllegalArgumentException("sink must declare at least one input topic");
}

static boolean hasInputs(SinkConfig cfg) {
    return cfg != null && ((cfg.getInputSpecs() != null && !cfg.getInputSpecs().isEmpty())
        || (cfg.getTopicsToSerdeClassName() != null && !cfg.getTopicsToSerdeClassName().isEmpty())
        || (cfg.getTopicToSchemaType() != null && !cfg.getTopicToSchemaType().isEmpty())
        || cfg.getTopicsPattern() != null);
}

try {
    SinkConfigUtils.validateAndExtractDetails(cfg, sinkPkg, transformPkg, true);
} catch (IllegalArgumentException e) {
    if (e.getMessage().startsWith("Must specify at least one topic")) {
        log.error("Sink has no input topics: set topicPattern/inputSpecs/topicsToSerdeClassName", e);
    }
    throw e;
}

When it happens

Trigger: SinkConfig with no topicPattern, no topicsToSerdeClassName, no topicToSchemaType, no inputSpecs, and no topicName — submitting via createSink/admin API.

Common situations: New sink configs created from templates where only the sink class and tenant/namespace/name were filled in; users confusing sink 'topicsPattern' with source 'topicsPattern'; config converter dropping input fields.

Understand the failure class

Background: "Missing required field" and "field is required" errors: why libraries reject payloads that omit mandatory fields — this error's family across 20 libraries.

Related errors

• Sink tenant cannot be null
• Sink namespace cannot be null
• Sink name cannot be null
• No input topic(s) specified for the function
• Input topic %s is invalid

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