{"record":{"id":"310b088f2ef414a2","repo":"apache/pulsar","slug":"receiver-queue-size-should-be-zero","errorCode":null,"errorMessage":"Receiver queue size should be >= zero","messagePattern":"Receiver queue size should be >= zero","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java","lineNumber":919,"sourceCode":"        }\n        if (!isEmpty(functionConfig.getGo()) && !org.apache.pulsar.common.functions.Utils\n                .isFunctionPackageUrlSupported(functionConfig.getGo())\n                && functionConfig.getGo().startsWith(BUILTIN)) {\n            String filename = functionConfig.getGo();\n            if (filename.contains(\"..\")) {\n                throw new IllegalArgumentException(\"Invalid filename: \" + filename);\n            }\n\n            if (!new File(filename).exists()) {\n                throw new IllegalArgumentException(\"The supplied go file does not exist\");\n            }\n        }\n\n        if (functionConfig.getInputSpecs() != null) {\n            functionConfig.getInputSpecs().forEach((topicName, conf) -> {\n                // receiver queue size should be >= 0\n                if (conf.getReceiverQueueSize() != null && conf.getReceiverQueueSize() < 0) {\n                    throw new IllegalArgumentException(\n                        \"Receiver queue size should be >= zero\");\n                }\n\n                if (conf.getCryptoConfig() != null && isBlank(conf.getCryptoConfig().getCryptoKeyReaderClassName())) {\n                    throw new IllegalArgumentException(\n                            \"CryptoKeyReader class name required\");\n                }\n                if (conf.getMessagePayloadProcessorConfig() != null && isBlank(\n                        conf.getMessagePayloadProcessorConfig().getClassName())) {\n                    throw new IllegalArgumentException(\n                            \"MessagePayloadProcessor class name required\");\n                }\n            });\n        }\n\n        if (functionConfig.getProducerConfig() != null\n                && functionConfig.getProducerConfig().getCryptoConfig() != null) {\n            if (isBlank(functionConfig.getProducerConfig().getCryptoConfig().getCryptoKeyReaderClassName())) {","sourceCodeStart":901,"sourceCodeEnd":937,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java#L901-L937","documentation":"Within each entry of FunctionConfig.inputSpecs (InputSpec map keyed by topic), receiverQueueSize must be zero or a positive number. A negative value cannot be honored by the consumer subscription and throws IllegalArgumentException('Receiver queue size should be >= zero') while iterating inputSpecs in doCommonChecks.","triggerScenarios":"Creating/updating a function whose InputSpec for any input topic has setReceiverQueueSize(-1) (or any negative Integer), e.g. conf.setInputSpecs(Map.of(\"topic\", new InputSpec().setReceiverQueueSize(-1))).","commonSituations":"Using -1 as a sentinel meaning 'unbounded/default' (not valid here); arithmetic or config-parsing bugs producing negative sizes; copying consumer configs where a different library accepted negatives; YAML/JSON edits that introduce a minus sign.","solutions":["Set receiverQueueSize to 0 or a positive value (e.g. 0 to disable buffering, 1000 for a larger buffer)","Remove the explicit receiverQueueSize so the default is used","Clamp/validate the value before building InputSpecs: Math.max(0, configuredSize)","Fix the upstream config source (YAML/JSON/env) that produced the negative number"],"exampleFix":"// before\nInputSpec spec = new InputSpec().setReceiverQueueSize(-1);\n// after\nInputSpec spec = new InputSpec().setReceiverQueueSize(1000); // must be >= 0","handlingStrategy":"validation","validationCode":"conf.getInputSpecs().forEach((topic, spec) -> {\n    Integer q = spec.getReceiverQueueSize();\n    if (q != null && q < 0) {\n        spec.setReceiverQueueSize(Math.max(0, q)); // or throw your own clear error\n    }\n});","typeGuard":"boolean hasValidReceiverQueueSizes(FunctionConfig c) {\n    return c.getInputSpecs() == null || c.getInputSpecs().values().stream()\n        .allMatch(s -> s.getReceiverQueueSize() == null || s.getReceiverQueueSize() >= 0);\n}","tryCatchPattern":"try {\n    admin.functions().createFunction(conf);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().contains(\"Receiver queue size\")) {\n        conf.getInputSpecs().values().forEach(s ->\n            { if (s.getReceiverQueueSize() != null && s.getReceiverQueueSize() < 0) s.setReceiverQueueSize(0); });\n        admin.functions().createFunction(conf);\n    } else throw e;\n}","preventionTips":["Clamp receiverQueueSize with Math.max(0, value) at config load time","Do not use -1 as a 'default/unbounded' sentinel in function InputSpecs","Validate deserialized YAML/JSON configs before submitting to the admin API"],"tags":["pulsar-functions","config-validation","illegal-argument"],"backgroundTag":"invalid-config-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"}