What causes "Invalid option %s. Must be a positive integer." in apache/flink?

Set the offending option to a positive value (>=1 for ints, >=1 for longs).

apache/flink · error · IllegalArgumentException

Invalid option %s. Must be a positive integer.

Error message

Invalid option %s. Must be a positive integer.

What it means

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.

Source

Thrown at flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/table/sink/options/AsyncSinkConfigurationValidator.java:94

                intVal -> intVal > 0,
                String.format("Invalid option %s. Must be a positive integer.", option.key()));
    }

    private void validatePositiveLongValue(ConfigOption<Long> option) {
        validateOptionValue(
                option,
                longVal -> longVal > 0L,
                String.format("Invalid option %s. Must be a positive integer.", option.key()));
    }

    private <T> void validateOptionValue(
            ConfigOption<T> option, Predicate<T> valueValidator, String errorMessage) {
        tableOptions
                .getOptional(option)
                .ifPresent(
                        val -> {
                            if (!valueValidator.test(val)) {
                                throw new IllegalArgumentException(errorMessage);
                            }
                        });
    }
}

View on GitHub (pinned to 2f3c205e92)

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.

Example fix

-- before
'sink.buffer-flush.max-rows' = '0',
-- after
'sink.buffer-flush.max-rows' = '100',

Defensive patterns

Strategy: validation

Validate before calling

// Validate async sink options before building the sink:
Map<String, Long> positive = Map.of(
    MAX_BATCH_SIZE.key(), 1L,
    MAX_BUFFERED_REQUESTS.key(), 1L,
    MAX_IN_FLIGHT_REQUESTS.key(), 1L);
for (Map.Entry<String, Long> e : positive.entrySet()) {
    int v = config.get(keyFromOption(e.getKey()));
    if (v <= 0) throw new IllegalArgumentException(e.getKey() + " must be > 0");
}

Try / catch

try {
    validator.getValidatedConfigurations();
} catch (IllegalArgumentException e) {
    // fix the offending option (name is in e.getMessage()) then retry
    throw e;
}

Prevention

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.

When it happens

Trigger: 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.

Common situations: Misconfigured async sink (e.g. setting 'max-buffered-requests' = 0 to 'disable buffering'); copy-paste errors; unit tests with placeholder values.

Related errors

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/36547644bcabf3e3. Report an issue: GitHub.