{"record":{"id":"ee3287ac5142733f","repo":"apache/pulsar","slug":"source-parallelism-must-be-a-positive-number","errorCode":null,"errorMessage":"Source parallelism must be a positive number","messagePattern":"Source parallelism must be a positive number","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java","lineNumber":284,"sourceCode":"            throw new IllegalArgumentException(\"Source tenant cannot be null\");\n        }\n        if (isEmpty(sourceConfig.getNamespace())) {\n            throw new IllegalArgumentException(\"Source namespace cannot be null\");\n        }\n        if (isEmpty(sourceConfig.getName())) {\n            throw new IllegalArgumentException(\"Source name cannot be null\");\n        }\n        if (!isEmpty(sourceConfig.getTopicName()) && !TopicName.isValid(sourceConfig.getTopicName())) {\n            throw new IllegalArgumentException(\"Topic name is invalid\");\n        }\n        if (!isEmpty(sourceConfig.getLogTopic())) {\n            if (!TopicName.isValid(sourceConfig.getLogTopic())) {\n                throw new IllegalArgumentException(\n                        String.format(\"LogTopic topic %s is invalid\", sourceConfig.getLogTopic()));\n            }\n        }\n        if (sourceConfig.getParallelism() != null && sourceConfig.getParallelism() <= 0) {\n            throw new IllegalArgumentException(\"Source parallelism must be a positive number\");\n        }\n        if (sourceConfig.getResources() != null) {\n            ResourceConfigUtils.validate(sourceConfig.getResources());\n        }\n\n        String sourceClassName = sourceConfig.getClassName();\n        // if class name in source config is not set, this should be a built-in source\n        // thus we should try to find it class name in the NAR service definition\n        if (sourceClassName == null) {\n            ConnectorDefinition connectorDefinition = sourceFunction.getFunctionMetaData(ConnectorDefinition.class);\n            if (connectorDefinition == null) {\n                throw new IllegalArgumentException(\n                        \"Source package doesn't contain the META-INF/services/pulsar-io.yaml file.\");\n            }\n            sourceClassName = connectorDefinition.getSourceClass();\n            if (sourceClassName == null) {\n                throw new IllegalArgumentException(\"Failed to extract source class from archive\");\n            }","sourceCodeStart":266,"sourceCodeEnd":302,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java#L266-L302","documentation":"validateAndExtractDetails checks that SourceConfig.parallelism, when provided, is strictly greater than zero. Parallelism controls how many instances of the source run; zero or a negative value is nonsensical, so an IllegalArgumentException is thrown. A null parallelism is allowed (the framework applies its own default).","triggerScenarios":"Submitting a source with sourceConfig.setParallelism(0) or a negative value — often from a CLI flag '--parallelism 0', an env-driven integer defaulting to 0, or arithmetic producing a non-positive count.","commonSituations":"Users passing 0 thinking it means 'unlimited' or 'auto'; parsing an empty/unset env var into 0 and forwarding it; computing parallelism from cluster size when the size query returned 0.","solutions":["Set parallelism to at least 1, e.g. sourceConfig.setParallelism(1)","If parallelism is optional, leave it null instead of 0 so the framework default applies","Guard numeric inputs before setting: only call setParallelism when the parsed value is > 0"],"exampleFix":"// before\nint p = Integer.parseInt(System.getenv(\"PARALLELISM\")); // 0 when unset\ncfg.setParallelism(p);\n// after\nString pRaw = System.getenv(\"PARALLELISM\");\nif (pRaw != null && Integer.parseInt(pRaw) > 0) {\n    cfg.setParallelism(Integer.parseInt(pRaw));\n}","handlingStrategy":"validation","validationCode":"if (cfg.getParallelism() != null && cfg.getParallelism() <= 0) {\n    throw new IllegalArgumentException(\"parallelism must be > 0 (or null for the default)\");\n}","typeGuard":"static boolean hasPositiveParallelism(SourceConfig cfg) {\n    return cfg == null || cfg.getParallelism() == null || cfg.getParallelism() > 0;\n}","tryCatchPattern":"try {\n    SourceConfigUtils.validateAndExtractDetails(cfg, pkg, true);\n} catch (IllegalArgumentException e) {\n    if (\"Source parallelism must be a positive number\".equals(e.getMessage())) {\n        // fall back to default parallelism or clamp the value to >= 1\n    }\n}","preventionTips":["Leave parallelism null unless you have an explicit instance count","Clamp env/CLI-derived integers: Math.max(1, parsedValue) when the field is present","Never map 'unset' inputs to 0 — map them to null"],"tags":["pulsar-functions","config-validation","illegal-argument"],"backgroundTag":"invalid-numeric-parameter","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"}