checkpoint.interval is set to

Error message

checkpoint.interval is set to {} which is not positive, fallback to default {} ms for streaming job.

What it means

AbstractFlinkRuntimeEnvironment.setCheckpoint validates the 'checkpoint.interval' config for streaming jobs. If the user set an explicit interval that is zero or negative (not a valid Flink checkpoint duration), it logs this warning and substitutes DEFAULT_CHECKPOINT_INTERVAL_MS instead of propagating the invalid value to Flink, which would throw at job construction. The job still runs with checkpointing enabled at the default interval.

Solutions

Set checkpoint.interval to a positive millisecond value (e.g. 60000) in the job config.
If checkpointing should be off, do not set checkpoint.interval at all (or rely on the documented disable path) instead of using 0.
Confirm effective behavior from the log line showing the fallback default interval and adjust if the default does not match your recovery-time requirements.

Example fix

// before
checkpoint.interval = 0
// after
checkpoint.interval = 60000

Defensive patterns

Strategy: validation

Validate before calling

// validate before submit
long interval = config.getLong("checkpoint.interval", -1);
if (config.hasPath("checkpoint.interval") && interval <= 0) {
    throw new IllegalArgumentException("checkpoint.interval must be > 0 ms, got: " + interval);
}

Prevention

Always configure checkpoint.interval as a positive millisecond value
Do not use 0 to disable checkpointing in streaming jobs — omit the option instead
Check startup logs for the fallback warning to confirm the effective interval

When it happens

Trigger: setCheckpoint() is invoked from createStreamEnvironment for a streaming job where checkpoint.interval was explicitly configured (hasExplicitInterval) but the value is <= 0 (positiveInterval == false).

Common situations: Config typo like `checkpoint.interval = 0` or a negative value; interval computed from a variable/placeholder that resolved to 0; misunderstanding that 0 disables checkpointing (it only does so when the option is unset or <=0 for batch; for streaming it is replaced by the default).

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/83e7a9d992b5ea0b. Report an issue: GitHub.

Appendix: source

Thrown at seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/execution/AbstractFlinkRuntimeEnvironment.java:98

    public StreamExecutionEnvironment getStreamExecutionEnvironment() {
        return environment;
    }

    protected void setCheckpoint() {
        OptionalLong intervalOpt = resolveCheckpointInterval(true);
        boolean hasExplicitInterval = intervalOpt.isPresent();
        boolean positiveInterval = intervalOpt.isPresent() && intervalOpt.getAsLong() > 0;
        long interval = intervalOpt.orElse(DEFAULT_CHECKPOINT_INTERVAL_MS);

        if (jobMode == JobMode.BATCH && !positiveInterval) {
            LOGGER.info(
                    "Checkpoint is disabled for batch job because 'checkpoint.interval' is not set or <= 0.");
            return;
        }

        if (hasExplicitInterval && !positiveInterval) {
            LOGGER.warn(
                    "checkpoint.interval is set to {} which is not positive, fallback to default {} ms for streaming job.",
                    interval,
                    DEFAULT_CHECKPOINT_INTERVAL_MS);
            interval = DEFAULT_CHECKPOINT_INTERVAL_MS;
        }

        CheckpointConfig checkpointConfig = environment.getCheckpointConfig();
        environment.enableCheckpointing(interval);

        if (config.hasPath(EnvCommonOptions.CHECKPOINT_TIMEOUT.key())) {
            long timeout = config.getLong(EnvCommonOptions.CHECKPOINT_TIMEOUT.key());
            checkpointConfig.setCheckpointTimeout(timeout);
        } else if (EnvironmentUtil.hasPathAndWaring(config, ConfigKeyName.CHECKPOINT_TIMEOUT)) {
            long timeout = config.getLong(ConfigKeyName.CHECKPOINT_TIMEOUT);
            checkpointConfig.setCheckpointTimeout(timeout);
        } else if (config.hasPath(EnvCommonOptions.CHECKPOINT_MIN_PAUSE.key())) {
            long minPause = config.getLong(EnvCommonOptions.CHECKPOINT_MIN_PAUSE.key());
            checkpointConfig.setMinPauseBetweenCheckpoints(minPause);

View on GitHub (pinned to cf67b549a7)