{"record":{"id":"60c077aedac4dedf","repo":"apache/pulsar","slug":"input-topic-s-is-invalid-60c077","errorCode":null,"errorMessage":"Input topic %s is invalid","messagePattern":"Input topic (.+?) is invalid","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java","lineNumber":436,"sourceCode":"        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\n        if (sinkConfig.getResources() != null) {\n            ResourceConfigUtils.validate(sinkConfig.getResources());\n        }\n\n        if (sinkConfig.getTimeoutMs() != null && sinkConfig.getTimeoutMs() < 0) {","sourceCodeStart":418,"sourceCodeEnd":454,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java#L418-L454","documentation":"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.","triggerScenarios":"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.","commonSituations":"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).","solutions":["Use a fully-qualified topic name: persistent://<tenant>/<namespace>/<topic>.","Trim whitespace and re-check for typos/slashes in the config value.","Validate locally with TopicName.isValid(topic) before submitting.","If using a pattern input, verify the pattern is a valid topic-pattern string."],"exampleFix":"// before\ninputSpecs.put(\"my-topic\", new ConsumerConfig());\n// after\ninputSpecs.put(\"persistent://public/default/my-topic\", new ConsumerConfig());","handlingStrategy":"validation","validationCode":"for (String topic : collectInputs(sinkConfig)) {\n    if (!org.apache.pulsar.common.naming.TopicName.isValid(topic)) {\n        throw new IllegalArgumentException(\"invalid input topic: \" + topic);\n    }\n}","typeGuard":"static boolean isValidTopic(String topic) {\n    return topic != null && org.apache.pulsar.common.naming.TopicName.isValid(topic.trim());\n}","tryCatchPattern":"try {\n    SinkConfigUtils.validateAndExtractDetails(cfg, sinkPkg, transformPkg, true);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().startsWith(\"Input topic\")) {\n        log.error(\"Malformed input topic in config: {}\", e.getMessage(), e);\n    }\n    throw e;\n}","preventionTips":["Always use fully-qualified topic names: persistent://tenant/namespace/topic.","Trim and sanitize topic strings read from YAML/CLI args.","Validate each topic with TopicName.isValid before submission.","Quote topic values in YAML to avoid parser coercion."],"tags":["pulsar","config-validation","topic-name"],"backgroundTag":"invalid-topic-name","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}