[SPARK-59635] Requeue immediately when AppRunningStep changes the running state

### What changes were proposed in this pull request?

This PR aims to make `AppRunningStep` requeue immediately (`completeAndImmediateRequeue()`) instead of using the default interval (`completeAndDefaultRequeue()`) when it appends a new running state (e.g. `DriverReady -> RunningHealthy`).

The next reconcile then takes the `proposedStateSummary == prevStateSummary` branch and observes the driver with `AppDriverRunningObserver`. This is the same way `AppReconcileStep.observeDriver` already works when it changes the state.

### Why are the changes needed?

When `AppRunningStep` changes the state, it completes the reconcile loop, so the same reconcile doesn't observe driver termination. The following `AppResourceObserveStep(AppDriverRunningObserver)` doesn't run either. If no more pod events arrive, the termination is only detected by the periodic reconcile (`spark.kubernetes.operator.reconciler.intervalSeconds`).

This happens easily with the per-resource reconcile rate limiter (`refreshPeriodSeconds=15`, `maxLoopForPeriod=5` by default). A short job's driver and executor pod events are postponed to the end of the rate-limit period. The postponed reconcile then sees `DriverReady` while the driver pod has already succeeded. For example, `tests/e2e/state-transition` failed locally because `Succeeded` was recorded 120 seconds after `RunningHealthy`, which is longer than the 60s assert timeout:

```
14:17:50.09 Submitted
14:17:50.39 DriverRequested
14:17:50.86 DriverStarted / DriverReady
14:18:05.09 RunningHealthy
14:20:05.11 Succeeded
```

With this PR, `Succeeded` is recorded right after `RunningHealthy`:

```
14:38:14.650 RunningHealthy
14:38:14.660 Succeeded
14:38:14.762 ResourceReleased
```

The cost is one extra reconcile for each running state transition.

### Does this PR introduce _any_ user-facing change?

Yes. The operator now detects driver termination right after it reports a running state, instead of waiting for the next periodic reconcile.

### How was this patch tested?

Pass the CIs with the newly added test case in `AppRunningStepTest`.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Opus 5

Closes #845 from dongjoon-hyun/SPARK-59635.

Authored-by: Dongjoon Hyun <dongjoon@apache.org>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: dongjoon-hyun

spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppRunningStep.java

@@ -20,7 +20,7 @@
 package org.apache.spark.k8s.operator.reconciler.reconcilesteps;
 
 import static org.apache.spark.k8s.operator.Constants.*;
-import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndDefaultRequeue;
+import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndImmediateRequeue;
 
 import java.util.List;
 import java.util.Set;
@@ -107,7 +107,7 @@ public ReconcileProgress reconcile(
               .getStatus()
               .appendNewState(new ApplicationState(proposedStateSummary, stateMessage));
       return attemptStatusUpdate(
-          context, statusRecorder, updatedStatus, completeAndDefaultRequeue());
+          context, statusRecorder, updatedStatus, completeAndImmediateRequeue());
     }
   }
 

spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppRunningStepTest.java

@@ -37,6 +37,7 @@
 
 import org.apache.spark.k8s.operator.SparkApplication;
 import org.apache.spark.k8s.operator.context.SparkAppContext;
+import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
 import org.apache.spark.k8s.operator.spec.ApplicationSpec;
 import org.apache.spark.k8s.operator.spec.ApplicationTolerations;
 import org.apache.spark.k8s.operator.spec.ExecutorInstanceConfig;
@@ -164,6 +165,24 @@ void runningHealthyDynamicAllocationEnabled() {
     assertEquals(ApplicationStateSummary.RunningHealthy, currentState);
   }
 
+  @Test
+  void immediateRequeueWhenRunningStateChanges() {
+    // Requeue immediately so that the next reconcile observes the driver in the new state
+    appSpec.setApplicationTolerations(ApplicationTolerations.builder().build());
+
+    appStatus =
+        appStatus.appendNewState(
+            new ApplicationState(ApplicationStateSummary.DriverReady, "Driver ready"));
+    app.setStatus(appStatus);
+
+    ReconcileProgress progress = appRunningStep.reconcile(mockContext, mockRecorder);
+
+    assertEquals(
+        ApplicationStateSummary.RunningHealthy,
+        app.getStatus().getCurrentState().getCurrentStateSummary());
+    assertEquals(ReconcileProgress.completeAndImmediateRequeue(), progress);
+  }
+
   @Test
   void runningHealthyExecutorsAtMaxCapacity() {
     // Dynamic allocation disabled, executors at max capacity