{"record":{"id":"4c877bf54cc9f4c8","repo":"apache/pulsar","slug":"must-specify-at-least-one-topic-of-input-via-topic","errorCode":null,"errorMessage":"Must specify at least one topic of input via topicToSerdeClassName, topicsPattern, topicToSchemaType or inputSpecs","messagePattern":"Must specify at least one topic of input via topicToSerdeClassName, topicsPattern, topicToSchemaType or inputSpecs","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java","lineNumber":431,"sourceCode":"\n    public static ExtractedSinkDetails validateAndExtractDetails(SinkConfig sinkConfig,\n                                                                 ValidatableFunctionPackage sinkFunction,\n                                                                 ValidatableFunctionPackage transformFunction,\n                                                                 boolean validateConnectorConfig) {\n        if (isEmpty(sinkConfig.getTenant())) {\n            throw new IllegalArgumentException(\"Sink tenant cannot be null\");\n        }\n        if (isEmpty(sinkConfig.getNamespace())) {\n            throw new IllegalArgumentException(\"Sink namespace cannot be null\");\n        }\n        if (isEmpty(sinkConfig.getName())) {\n            throw new IllegalArgumentException(\"Sink name cannot be null\");\n        }\n\n        // make we sure we have one source of input\n        Collection<String> allInputs = collectAllInputTopics(sinkConfig);\n        if (allInputs.isEmpty()) {\n            throw new IllegalArgumentException(\"Must specify at least one topic of input via topicToSerdeClassName, \"\n                    + \"topicsPattern, topicToSchemaType or inputSpecs\");\n        }\n        for (String topic : allInputs) {\n            if (!TopicName.isValid(topic)) {\n                throw new IllegalArgumentException(String.format(\"Input topic %s is invalid\", topic));\n            }\n        }\n        if (!isEmpty(sinkConfig.getLogTopic())) {\n            if (!TopicName.isValid(sinkConfig.getLogTopic())) {\n                throw new IllegalArgumentException(\n                        String.format(\"LogTopic topic %s is invalid\", sinkConfig.getLogTopic()));\n            }\n        }\n\n        if (sinkConfig.getParallelism() != null && sinkConfig.getParallelism() <= 0) {\n            throw new IllegalArgumentException(\"Sink parallelism must be a positive number\");\n        }\n","sourceCodeStart":413,"sourceCodeEnd":449,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java#L413-L449","documentation":"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.","triggerScenarios":"SinkConfig with no topicPattern, no topicsToSerdeClassName, no topicToSchemaType, no inputSpecs, and no topicName — submitting via createSink/admin API.","commonSituations":"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.","solutions":["Set at least one input: cfg.setTopicPattern(\"persistent://public/default/my-topic\") or cfg.setInputSpecs(Map.of(\"my-topic\", new ConsumerConfig())).","Or set topicsToSerdeClassName / topicToSchemaType with the source topic(s).","For a single-topic sink, set topicName plus its serde/schema config.","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."],"exampleFix":"// before\nSinkConfig cfg = new SinkConfig();\ncfg.setTenant(\"public\");\ncfg.setNamespace(\"default\");\ncfg.setName(\"my-sink\");\ncfg.setClassName(\"org.example.MySink\");\n// after\nSinkConfig cfg = new SinkConfig();\ncfg.setTenant(\"public\");\ncfg.setNamespace(\"default\");\ncfg.setName(\"my-sink\");\ncfg.setClassName(\"org.example.MySink\");\nMap<String, ConsumerConfig> inputSpecs = new HashMap<>();\ninputSpecs.put(\"persistent://public/default/my-topic\", new ConsumerConfig());\ncfg.setInputSpecs(inputSpecs);","handlingStrategy":"validation","validationCode":"boolean hasInputs = sinkConfig.getInputSpecs() != null && !sinkConfig.getInputSpecs().isEmpty()\n    || (sinkConfig.getTopicsToSerdeClassName() != null && !sinkConfig.getTopicsToSerdeClassName().isEmpty())\n    || (sinkConfig.getTopicToSchemaType() != null && !sinkConfig.getTopicToSchemaType().isEmpty())\n    || sinkConfig.getTopicsPattern() != null;\nif (!hasInputs) {\n    throw new IllegalArgumentException(\"sink must declare at least one input topic\");\n}","typeGuard":"static boolean hasInputs(SinkConfig cfg) {\n    return cfg != null && ((cfg.getInputSpecs() != null && !cfg.getInputSpecs().isEmpty())\n        || (cfg.getTopicsToSerdeClassName() != null && !cfg.getTopicsToSerdeClassName().isEmpty())\n        || (cfg.getTopicToSchemaType() != null && !cfg.getTopicToSchemaType().isEmpty())\n        || cfg.getTopicsPattern() != null);\n}","tryCatchPattern":"try {\n    SinkConfigUtils.validateAndExtractDetails(cfg, sinkPkg, transformPkg, true);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().startsWith(\"Must specify at least one topic\")) {\n        log.error(\"Sink has no input topics: set topicPattern/inputSpecs/topicsToSerdeClassName\", e);\n    }\n    throw e;\n}","preventionTips":["Always configure inputSpecs or topicsPattern when creating a sink.","Do not confuse source output specs with sink input specs.","Keep a minimal working sink config as a template.","Diff generated configs against a known-good example before submitting."],"tags":["pulsar","config-validation","sink","topics"],"backgroundTag":"missing-required-field","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}