From f66e43d4c437df63c60c72c728c5abbb23c81e07 Mon Sep 17 00:00:00 2001
From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com>
Date: Fri, 3 Jul 2026 21:31:15 +0200
Subject: [PATCH] [FLINK-40068][tests] Do not assert apply() was called when
 cancelling KeyedJob window

StatefulWindowFunction.close() also runs on the cancellation path, and the GENERATE/MIGRATE jobs are always stopped via non-draining cancel-with-savepoint, so a window subtask can be closed before apply() was called; the failing assertion then kills the shared MiniCluster's only TaskManager and starves every later parameterization. Restrict the assertion to RESTORE, the only mode whose job runs to completion.

Generated-by: Claude Opus 4.8 (1M context)
---
 .../flink/test/state/operator/restore/keyed/KeyedJob.java   | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/keyed/KeyedJob.java b/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/keyed/KeyedJob.java
index 58a47f5bafa29..1713db4e78a2d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/keyed/KeyedJob.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/keyed/KeyedJob.java
@@ -214,7 +214,11 @@ public void apply(
 
         @Override
         public void close() {
-            assertThat(applyCalled).as("Apply was never called.").isTrue();
+            // GENERATE/MIGRATE stop via non-draining cancel-with-savepoint, so a subtask may
+            // close before apply() ran; only RESTORE runs to completion.
+            if (mode == ExecutionMode.RESTORE) {
+                assertThat(applyCalled).as("Apply was never called.").isTrue();
+            }
         }
     }