{"record":{"id":"8cf7bd7a923d1d77","repo":"apache/pulsar","slug":"sink-timeout-must-be-a-positive-number","errorCode":null,"errorMessage":"Sink timeout must be a positive number","messagePattern":"Sink timeout 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":455,"sourceCode":"            }\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.\");\n            }\n            sinkClassName = connectorDefinition.getSinkClass();\n            if (sinkClassName == null) {\n                throw new IllegalArgumentException(\"Failed to extract sink class from archive\");\n            }\n        }\n\n        // check if sink implements the correct interfaces","sourceCodeStart":437,"sourceCodeEnd":473,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java#L437-L473","documentation":"If the sink config sets a timeoutMs, it must be non-negative (the message says 'positive', the check rejects negative values; null means no timeout). validateAndExtractDetails throws IllegalArgumentException when timeoutMs < 0.","triggerScenarios":"SinkConfig.setTimeoutMs(-1) — often from using -1 as a sentinel 'disabled' value in code or config, or an arithmetic underflow when computing the timeout.","commonSituations":"Users setting '-1' to disable ack timeout based on other systems' conventions; duration parsing producing negative values; copy-pasted config from tools where -1 means infinite.","solutions":["Set a non-negative timeout in milliseconds, e.g. sinkConfig.setTimeoutMs(30000L).","To disable the timeout, leave timeoutMs unset (null) instead of -1.","Fix duration computation that underflows to a negative value.","Validate timeout >= 0 before submitting."],"exampleFix":"// before\nsinkConfig.setTimeoutMs(-1L); // intended 'disabled'\n// after\n// leave unset to disable:\n// sinkConfig.setTimeoutMs(null); — or set a real value:\nsinkConfig.setTimeoutMs(30000L);","handlingStrategy":"validation","validationCode":"if (sinkConfig.getTimeoutMs() != null && sinkConfig.getTimeoutMs() < 0) {\n    throw new IllegalArgumentException(\"timeoutMs must be >= 0\");\n}","typeGuard":"static boolean hasValidTimeout(SinkConfig cfg) {\n    return cfg.getTimeoutMs() == null || cfg.getTimeoutMs() >= 0;\n}","tryCatchPattern":"try {\n    SinkConfigUtils.validateAndExtractDetails(cfg, sinkPkg, transformPkg, true);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().contains(\"timeout\")) {\n        log.error(\"Negative sink timeoutMs: {}\", cfg.getTimeoutMs(), e);\n    }\n    throw e;\n}","preventionTips":["Treat null as 'no timeout' — don't use -1 as a sentinel.","Check duration arithmetic for underflow before assigning timeoutMs.","Validate timeouts read from config files before submission.","Document timeout semantics (ms, null = disabled) in your tooling."],"tags":["pulsar","config-validation","timeout"],"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"}