This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new e206faa4ca2 [FLINK-33186][tests] Re-disable periodic checkpoints in
CheckpointAfterAllTasksFinishedITCase tests (#29517)
e206faa4ca2 is described below
commit e206faa4ca2dbd174c4bf7615db1f216cbf591ce
Author: Efrat Levitan <[email protected]>
AuthorDate: Fri Oct 23 09:46:25 2026 +0300
[FLINK-33186][tests] Re-disable periodic checkpoints in
CheckpointAfterAllTasksFinishedITCase tests (#29517)
A few tests were originally meant to enable checkpointing with a large
interval, so no checkpoint is effectively triggered. Since FLINK-38939
the coordinator re-schedules the periodic trigger based on min pause.
With the default min pause of 0 a periodic checkpoint fires as soon as
all tasks are running.
Set the min pause of both tests to one year, the maximum the
coordinator accepts. The pause gates periodic triggers only, so the
end-of-data checkpoint still fires at once, while every scheduler start
or restart now waits longer than the test can run. All tasks then wait
for the same final checkpoint and none can finish before it is taken,
which removes the race instead of recovering from it.
Generated-by: Claude Code (Claude Fable 5.1)
---
.../CheckpointAfterAllTasksFinishedITCase.java | 17 ++++++++++-------
1 file changed, 10 insertions(+), 7 deletions(-)
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointAfterAllTasksFinishedITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointAfterAllTasksFinishedITCase.java
index badf21f736d..8bcfeee1ffa 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointAfterAllTasksFinishedITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointAfterAllTasksFinishedITCase.java
@@ -80,12 +80,18 @@ class CheckpointAfterAllTasksFinishedITCase extends
AbstractTestBase {
IntegerStreamSource.latch = new CountDownLatch(1);
}
+ private void disablePeriodicCheckpointing() {
+ env.enableCheckpointing(
+ Duration.ofNanos(Long.MAX_VALUE) /* max duration allowed by
FLINK */.toMillis());
+ // Since FLINK-38939, a large checkpointing interval has to be
combined with a large
+ // min-pause to effectively disable periodic checkpoints.
+
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(Duration.ofDays(365).toMillis());
+ }
+
@Test
void testImmediateCheckpointing() throws Exception {
RestartStrategyUtils.configureNoRestartStrategy(env);
- // Checkpointing is enabled with a large interval, and no checkpoints
will be triggered.
- env.enableCheckpointing(
- Duration.ofNanos(Long.MAX_VALUE) /* max duration allowed by
FLINK */.toMillis());
+ disablePeriodicCheckpointing();
StreamGraph streamGraph = getStreamGraph(env, false, false);
env.execute(streamGraph);
assertThat(smallResult.get().size()).isEqualTo(SMALL_SOURCE_NUM_RECORDS);
@@ -124,10 +130,7 @@ class CheckpointAfterAllTasksFinishedITCase extends
AbstractTestBase {
.get();
bigResult.get().clear();
- // Checkpointing is enabled with a large interval, and no
checkpoints will be triggered.
- env.enableCheckpointing(
- Duration.ofNanos(Long.MAX_VALUE) /* max duration allowed
by FLINK */
- .toMillis());
+ disablePeriodicCheckpointing();
JobGraph restoredJobGraph = getStreamGraph(env, true,
false).getJobGraph();
restoredJobGraph.setSavepointRestoreSettings(
SavepointRestoreSettings.forPath(savepointPath, false));
