Repository navigation
[test][python] Use the registered checkpoint interval key in the Python e2e tests - #978
Merged
Merged
Conversation
The Python e2e tests set `checkpointing.interval`, which is not a Flink configuration key. It is not registered, and it is not wired as a deprecated or fallback key either, so Flink ignored it silently. Because `execution.checkpointing.interval` carries `noDefaultValue()`, no periodic checkpointing was scheduled, and these tests configured a RocksDB state backend that was never snapshotted. Rename the key at all 11 call sites across 5 files. The neighbouring `state.backend.type` and `restart-strategy.type` are both registered options and are left unchanged. Verified against the shipped flink-dist jar rather than documentation: CheckpointingOptions#CHECKPOINTING_INTERVAL resolves to `execution.checkpointing.interval`, its fallback keys are empty, and reading the option from a Configuration holding the bare key returns null. Generated-by: Claude Code 2.1.224
Flink ignores an unregistered configuration key silently, with no exception and no log line, so a near miss such as `checkpointing.interval` leaves checkpointing disabled while the test still passes. No lint rule can catch it either, since the key is a plain string argument. Add a test that scans the e2e sources for the checkpoint interval keys they configure and compares them against CheckpointingOptions#CHECKPOINTING_INTERVAL read from Flink itself, so neither side of the comparison is a hand-written copy of the string. The pattern is deliberately loose on setter name, key stem, and quote style: a near miss Flink would silently ignore has to be caught whatever spelling it arrives in. The key is captured up to its closing quote, so a longer registered key such as `execution.checkpointing.interval-during-backlog` fails loudly rather than being truncated to a passing prefix. A scan that matches nothing also fails, so a stale pattern cannot pass vacuously. The test lives in the arm that runs the full Flink version matrix, since what it checks is whether the key resolves against a given Flink version. Generated-by: Claude Code 2.1.224
wenjin272
approved these changes
Aug 8, 2026
wenjin272
left a comment
Contributor
There was a problem hiding this comment.
Thanks for catching this. LGTM.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Linked issue: #976
Purpose of change
checkpointing.intervalis not a Flink configuration key. The real one isexecution.checkpointing.interval(CheckpointingOptions#CHECKPOINTING_INTERVAL). The bare form is not registered, and it is not wired as a deprecated or fallback key either, so Flink ignores it silently. Since the real option hasnoDefaultValue(), no periodic checkpointing was ever scheduled.The Python e2e tests set the bare form in 11 places across 5 files, so they configure a RocksDB state backend that is never snapshotted while reading as though they exercise checkpointing.
The first commit renames the key at all 11 sites. The neighbouring
state.backend.typeandrestart-strategy.typeare both genuine registered options and are left unchanged.The second commit adds one test so this cannot recur. Nothing today can catch this class of bug: Flink emits no signal for an unregistered key,
Configurationlogs only for registered fallbacks, PyFlink writes to a file appender rather than the console, and no lint rule can check a plain string argument. The guard scans the e2e sources for the checkpoint interval keys they configure and compares them against the key read from Flink itself, so neither side of the comparison is a hand-written copy of the string. A guard carrying its own copy of the key would pass while every real call site stayed broken.Two things worth setting expectations on. The rename does not make these tests meaningfully exercise checkpointing: measured on Flink 2.2 it yields one completed checkpoint per job, of largely drained state, because the operator drains its in-flight keys before the snapshot lands. Getting a checkpoint to complete with in-flight state did not happen at intervals down to 10ms, with slow agents, or with larger inputs. Doing better needs a different test design rather than a configuration fix.
The other is that the fix activates a path that was previously dead. At every site
tolerable-failed-checkpointsis 0 andrestart-strategy.typeisdisable, and with no checkpoint directory configured the storage falls back to JobManager memory, which caps state at 5 MiB. So a failed or oversized checkpoint now fails the job outright, where before it could not occur at all. Nothing hit this in practice, but it is the first thing to check if these tests ever go flaky.Tests
./tools/ut.sh -p -e -f 2.2passes. All 14 tests across the 5 renamed files pass, and they pass on the first attempt rather than being retried by--reruns.mcp_test.pyis normally skipped without a local Ollama, so a localqwen3:1.7bwas stood up to cover the 11th site rather than leaving it to CI. Both parametrizations pass.The new key was checked against the shipped
flink-distjar rather than taken from documentation:CHECKPOINTING_INTERVAL.key()resolves toexecution.checkpointing.interval, its fallback keys are empty, reading the option from aConfigurationholding the bare key returnsnull, and"1s"parses toPT1Sfor itsDurationtype, so the value needed no change. The same holds on 1.20.3. The option is present inflink-coreon 1.20, 2.0, 2.1, 2.2 and 2.3; 2.0, 2.1 and 2.3 were verified by inspection rather than executed locally.The guard was verified to fail, not just to pass. It goes red on the original defect, on a
set_integerspelling, onexecution.checkpoint.interval, on a single-quoted key, on all of those combined, on the longer registered keyexecution.checkpointing.interval-during-backlog, and on a scan that matches nothing. Each failure names the offending file. It produces no false positives: exactly 11 matches across the 5 files, resolving to the single expected key.The guard sits in the pytest selection that runs the full Flink version matrix, since what it checks is whether the key resolves against a given Flink version. It costs about 10ms there, where the JVM gateway is already warm.
API
No public API change. Both commits are test-only.
Documentation
doc-neededdoc-not-neededdoc-includedWas this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code 2.1.224