{"record":{"id":"02e64d2d876ac235","repo":"apache/pulsar","slug":"sink-parallelism-must-be-a-positive-number","errorCode":null,"errorMessage":"Sink parallelism must be a positive number","messagePattern":"Sink 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/SinkConfigUtils.java","lineNumber":447,"sourceCode":"        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\n        // thus we should try to find it class name in the NAR service definition\n        if (sinkClassName == null) {\n            ConnectorDefinition connectorDefinition = sinkFunction.getFunctionMetaData(ConnectorDefinition.class);\n            if (connectorDefinition == null) {\n                throw new IllegalArgumentException(\n                        \"Sink package doesn't contain the META-INF/services/pulsar-io.yaml file.\");","sourceCodeStart":429,"sourceCodeEnd":465,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java#L429-L465","documentation":"When the sink config specifies a parallelism, it must be strictly positive. validateAndExtractDetails throws IllegalArgumentException if parallelism is null-safe-checked but <= 0 (i.e. 0 or negative values passed explicitly).","triggerScenarios":"SinkConfig.setParallelism(0) or negative, e.g. a computed parallelism that evaluated to 0, or YAML 'parallelism: 0'.","commonSituations":"Dynamic parallelism computed from partition count of 0-partitioned/empty clusters; default-value bugs where an int field initializes to 0 instead of null; user setting 0 thinking it means 'auto'.","solutions":["Set parallelism to at least 1, e.g. sinkConfig.setParallelism(1).","Fix the code computing parallelism to clamp: Math.max(1, computed).","Leave parallelism unset (null) to let defaults apply.","Verify your YAML doesn't have 'parallelism: 0'."],"exampleFix":"// before\nsinkConfig.setParallelism(0);\n// after\nsinkConfig.setParallelism(1);","handlingStrategy":"validation","validationCode":"if (sinkConfig.getParallelism() != null && sinkConfig.getParallelism() <= 0) {\n    throw new IllegalArgumentException(\"parallelism must be >= 1\");\n}","typeGuard":"static boolean hasValidParallelism(SinkConfig cfg) {\n    return cfg.getParallelism() == null || cfg.getParallelism() > 0;\n}","tryCatchPattern":"try {\n    SinkConfigUtils.validateAndExtractDetails(cfg, sinkPkg, transformPkg, true);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().contains(\"parallelism\")) {\n        log.error(\"Sink parallelism must be positive: {}\", cfg.getParallelism(), e);\n    }\n    throw e;\n}","preventionTips":["Clamp computed parallelism with Math.max(1, value).","Never use 0 to mean 'auto' — leave it null or use 1.","Sanity-check YAML numeric fields parse to the intended values.","Default to parallelism 1 in config templates."],"tags":["pulsar","config-validation","parallelism"],"backgroundTag":"invalid-parameter-value","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"}