{"record":{"id":"0c9c7a3e883dc21b","repo":"apache/pulsar","slug":"logtopic-topic-s-is-invalid-0c9c7a","errorCode":null,"errorMessage":"LogTopic topic %s is invalid","messagePattern":"LogTopic 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":441,"sourceCode":"        }\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) {\n            throw new IllegalArgumentException(\"Sink timeout must be a positive number\");\n        }\n\n        String sinkClassName = sinkConfig.getClassName();\n        // if class name in sink config is not set, this should be a built-in sink","sourceCodeStart":423,"sourceCodeEnd":459,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java#L423-L459","documentation":"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.","triggerScenarios":"SinkConfig.setLogTopic(\"bad name\") or a logTopic YAML value that isn't a valid topic (missing domain prefix, invalid characters).","commonSituations":"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.","solutions":["Set a fully-qualified log topic: persistent://public/default/sink-logs.","Remove the logTopic field if logging to a topic is not needed.","Validate with TopicName.isValid before submission.","Check YAML quoting — unquoted values with special characters can corrupt the string."],"exampleFix":"// before\nsinkConfig.setLogTopic(\"sink-logs\");\n// after\nsinkConfig.setLogTopic(\"persistent://public/default/sink-logs\");","handlingStrategy":"validation","validationCode":"if (sinkConfig.getLogTopic() != null\n    && !org.apache.pulsar.common.naming.TopicName.isValid(sinkConfig.getLogTopic())) {\n    throw new IllegalArgumentException(\"invalid logTopic: \" + sinkConfig.getLogTopic());\n}","typeGuard":"static boolean hasValidLogTopic(SinkConfig cfg) {\n    return cfg.getLogTopic() == null\n        || org.apache.pulsar.common.naming.TopicName.isValid(cfg.getLogTopic().trim());\n}","tryCatchPattern":"try {\n    SinkConfigUtils.validateAndExtractDetails(cfg, sinkPkg, transformPkg, true);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().startsWith(\"LogTopic\")) {\n        log.error(\"Malformed logTopic in sink config: {}\", e.getMessage(), e);\n    }\n    throw e;\n}","preventionTips":["Use a fully-qualified topic for logTopic or omit it entirely.","Quote YAML values containing special characters.","Validate logTopic with TopicName.isValid when present.","Share a single constant for the log topic across configs."],"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-14T05:17:10.506Z"}