From b2aec64b63369b1086bc8cb1cb33d468a1cb4772 Mon Sep 17 00:00:00 2001
From: Roman Khachatryan <khachatryan.roman@gmail.com>
Date: Thu, 30 Jul 2026 14:52:35 +0200
Subject: [PATCH] [hotfix][runtime] Fix flaky NPE in SpeculativeExecutionTest

testCancelOtherScheduledCurrentExecutionsWhenAnyExecutionFinished starts
scheduling asynchronously (createSchedulerAndStartScheduling(false)) and
polls for the execution vertex via waitUntilIgnoringExceptions. That poll
loop only guards against exceptions thrown while fetching the vertex; it
discards the fetched value and unconditionally returns true, then the
vertex is fetched a second, independent time right after the loop. Between
these two fetches, the scheduler thread can still be populating the
execution graph, so the second fetch occasionally observes a null element
instead of throwing, causing:

NullPointerException: Cannot invoke
"ExecutionVertex.getCurrentExecutionAttempt()" because "ev" is null

Capture the vertex from the successful poll iteration itself, and make the
poll condition require a non-null result, instead of re-querying the
scheduler afterwards.
---
 .../SpeculativeExecutionTest.java             | 24 +++++++++++++++++++
 1 file changed, 24 insertions(+)

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/SpeculativeExecutionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/SpeculativeExecutionTest.java
index 46bee0254eb71..75807295506b2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/SpeculativeExecutionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/SpeculativeExecutionTest.java
@@ -74,6 +74,11 @@
 import java.util.Set;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+<<<<<<< Updated upstream
+=======
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+>>>>>>> Stashed changes
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -251,8 +256,27 @@ void testCancelOtherDeployedCurrentExecutionsWhenAnyExecutionFinished() {
     void testCancelOtherScheduledCurrentExecutionsWhenAnyExecutionFinished() {
         testExecutionSlotAllocator.disableAutoCompletePendingRequests();
 
+<<<<<<< Updated upstream
         final AdaptiveBatchScheduler scheduler = createSchedulerAndStartScheduling();
         final ExecutionVertex ev = getOnlyExecutionVertex(scheduler);
+=======
+        final AdaptiveBatchScheduler scheduler = createSchedulerAndStartScheduling(false);
+
+        // Scheduling started asynchronously (waitForDeployment=false), so the execution vertex
+        // may not exist yet. Capture it from the successful check itself instead of discarding
+        // it and querying again below - a second, independent query can otherwise race with the
+        // scheduler thread still populating the execution graph and observe a null element.
+        final AtomicReference<ExecutionVertex> evRef = new AtomicReference<>();
+        CommonTestUtils.waitUntilIgnoringExceptions(
+                () -> {
+                    evRef.set(getOnlyExecutionVertex(scheduler));
+                    return evRef.get() != null;
+                },
+                Duration.ofSeconds(1),
+                Duration.ofMillis(10),
+                "Failed to get the execution vertex.");
+        final ExecutionVertex ev = evRef.get();
+>>>>>>> Stashed changes
         final Execution attempt1 = ev.getCurrentExecutionAttempt();
 
         testExecutionSlotAllocator.completePendingRequest(attempt1.getAttemptId());