Prevent StepExecutionIterator from leaking memory in cases where a single processed execution has a stuck CPS VM thread

src/main/java/org/jenkinsci/plugins/workflow/flow/FlowExecutionList.java

@@ -253,24 +253,26 @@
 
             for (FlowExecution e : FlowExecutionList.get()) {
                 ListenableFuture<List<StepExecution>> execs = e.getCurrentExecutions(false);
-                all.add(execs);
-                Futures.addCallback(execs,new FutureCallback<List<StepExecution>>() {
-                    @Override
-                    public void onSuccess(@NonNull List<StepExecution> result) {
-                        for (StepExecution e : result) {
-                            try {
-                                f.apply(e);
-                            } catch (RuntimeException x) {
-                                LOGGER.log(Level.WARNING, null, x);
-                            }
+                // We transform the futures that return List<StepExecution> into futures that return Void before
+                // passing them to Futures.allAsList so that the combined future only holds strong references to each
+                // FlowExecution until its StepExecutions have been loaded and applied to the function. This prevents
+                // us from leaking references to all processed executions in the case where a single build has a stuck
+                // CpsVmExecutorService that prevents the future returned by getCurrentExecutions from completing.
+                ListenableFuture<Void> results = Futures.transform(execs, (List<StepExecution> result) -> {
+                    for (StepExecution se : result) {
+                        try {
+                            f.apply(se);
+                        } catch (RuntimeException x) {
+                            LOGGER.log(Level.WARNING, null, x);
                         }
                     }
-
-                    @Override
-                    public void onFailure(@NonNull Throwable t) {
-                        LOGGER.log(Level.WARNING, null, t);
-                    }
+                    return null;
+                }, MoreExecutors.directExecutor());
+                ListenableFuture<Void> resultsWithWarningsLogged = Futures.catching(results, Throwable.class, t -> {
+                    LOGGER.log(Level.WARNING, null, t);
+                    return null;
                 }, MoreExecutors.directExecutor());
+                all.add(resultsWithWarningsLogged);
             }
 
             return Futures.allAsList(all);

src/test/java/org/jenkinsci/plugins/workflow/flow/FlowExecutionListTest.java

@@ -38,12 +38,14 @@
 import hudson.model.TaskListener;
 import hudson.model.queue.QueueTaskFuture;
 import java.io.Serializable;
+import java.lang.ref.WeakReference;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Collections;
 import java.util.Set;
 import java.util.function.Supplier;
 import java.util.logging.Level;
+import jenkins.model.Jenkins;
 import org.hamcrest.Matcher;
 import org.jenkinsci.plugins.workflow.cps.CpsFlowDefinition;
 import org.jenkinsci.plugins.workflow.job.WorkflowJob;
@@ -60,6 +62,7 @@
 import org.jvnet.hudson.test.Issue;
 import org.jvnet.hudson.test.LoggerRule;
 import org.jvnet.hudson.test.JenkinsSessionRule;
+import org.jvnet.hudson.test.MemoryAssert;
 import org.jvnet.hudson.test.TestExtension;
 import org.kohsuke.stapler.DataBoundConstructor;
 
@@ -160,6 +163,32 @@
         });
     }
 
+    @Test public void stepExecutionIteratorDoesNotLeakBuildsWhenOneIsStuck() throws Throwable {
+        sessions.then(r -> {
+            var notStuck = r.createProject(WorkflowJob.class, "not-stuck");
+            notStuck.setDefinition(new CpsFlowDefinition("semaphore('wait')", true));
+            var notStuckBuild = notStuck.scheduleBuild2(0).waitForStart();
+            SemaphoreStep.waitForStart("wait/1", notStuckBuild);
+            WeakReference<Object> notStuckBuildRef = new WeakReference<>(notStuckBuild);
+            // Create a Pipeline that runs a long-lived task on its CpsVmExecutorService, causing it to get stuck.
+            var stuck = r.createProject(WorkflowJob.class, "stuck");
+            stuck.setDefinition(new CpsFlowDefinition("echo 'test message'; Thread.sleep(Integer.MAX_VALUE)", false));
+            var stuckBuild = stuck.scheduleBuild2(0).waitForStart();
+            r.waitForMessage("test message", stuckBuild);
+            Thread.sleep(1000); // We need Thread.sleep to be running in the CpsVmExecutorService.
+            // Make FlowExecutionList$StepExecutionIteratorImpl.applyAll submit a task to the CpsVmExecutorService
+            // for stuck #1 that will never complete, so the resulting future will never complete.
+            StepExecution.applyAll(e -> null);
+            // Let notStuckBuild complete and check that it can be GC'd.
+            SemaphoreStep.success("wait/1", null);
+            r.waitForCompletion(notStuckBuild);
+            notStuckBuild = null; // Clear out the local variable in this thread.
+            Jenkins.get().getQueue().clearLeftItems(); // We don't want to wait 5 minutes.
+            MemoryAssert.assertGC(notStuckBuildRef, true);
+            // TODO: Test cleanup hangs for 1 minute in CpsFlowExecution.suspendAll because the checkpoint task can't run.
+        });
+    }
+
     public static class NonResumableStep extends Step implements Serializable {
         public static final long serialVersionUID = 1L;
         @DataBoundConstructor