{"record":{"id":"36547644bcabf3e3","repo":"apache/flink","slug":"invalid-option-s-must-be-a-positive-integer","errorCode":null,"errorMessage":"Invalid option %s. Must be a positive integer.","messagePattern":"Invalid option (.+?)\\. Must be a positive integer\\.","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/table/sink/options/AsyncSinkConfigurationValidator.java","lineNumber":94,"sourceCode":"                intVal -> intVal > 0,\n                String.format(\"Invalid option %s. Must be a positive integer.\", option.key()));\n    }\n\n    private void validatePositiveLongValue(ConfigOption<Long> option) {\n        validateOptionValue(\n                option,\n                longVal -> longVal > 0L,\n                String.format(\"Invalid option %s. Must be a positive integer.\", option.key()));\n    }\n\n    private <T> void validateOptionValue(\n            ConfigOption<T> option, Predicate<T> valueValidator, String errorMessage) {\n        tableOptions\n                .getOptional(option)\n                .ifPresent(\n                        val -> {\n                            if (!valueValidator.test(val)) {\n                                throw new IllegalArgumentException(errorMessage);\n                            }\n                        });\n    }\n}\n","sourceCodeStart":76,"sourceCodeEnd":99,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/table/sink/options/AsyncSinkConfigurationValidator.java#L76-L99","documentation":"AsyncSinkConfigurationValidator rejects any of MAX_BATCH_SIZE, FLUSH_BUFFER_SIZE, MAX_BUFFERED_REQUESTS, MAX_IN_FLIGHT_REQUESTS, or FLUSH_BUFFER_TIMEOUT whose value is not strictly greater than zero. validateOptionValue applies a predicate and throws IllegalArgumentException with the offending option key.","triggerScenarios":"Setting any async sink option listed above to 0 or a negative number in SQL table options or programmatic config. getValidatedConfigurations() runs all five checks.","commonSituations":"Misconfigured async sink (e.g. setting 'max-buffered-requests' = 0 to 'disable buffering'); copy-paste errors; unit tests with placeholder values.","solutions":["Set the offending option to a positive value (>=1 for ints, >=1 for longs).","Remove the option entirely to rely on the documented default, which is always positive.","Double-check units: FLUSH_BUFFER_SIZE is bytes, FLUSH_BUFFER_TIMEOUT is ms.","Re-run getValidatedConfigurations to confirm no other option also fails."],"exampleFix":"-- before\n'sink.buffer-flush.max-rows' = '0',\n-- after\n'sink.buffer-flush.max-rows' = '100',","handlingStrategy":"validation","validationCode":"// Validate async sink options before building the sink:\nMap<String, Long> positive = Map.of(\n    MAX_BATCH_SIZE.key(), 1L,\n    MAX_BUFFERED_REQUESTS.key(), 1L,\n    MAX_IN_FLIGHT_REQUESTS.key(), 1L);\nfor (Map.Entry<String, Long> e : positive.entrySet()) {\n    int v = config.get(keyFromOption(e.getKey()));\n    if (v <= 0) throw new IllegalArgumentException(e.getKey() + \" must be > 0\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    validator.getValidatedConfigurations();\n} catch (IllegalArgumentException e) {\n    // fix the offending option (name is in e.getMessage()) then retry\n    throw e;\n}","preventionTips":["Always use positive integers for the five async sink options.","Omit options you do not intend to override so defaults apply.","Validate config once at table construction, not per record."],"tags":["config-validation","async-sink","options","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}