[SPARK-59576] Support K8s Normal events for all SparkApplication and SparkCluster state transitions

[dongjoon-hyun]

What changes were proposed in this pull request?

This PR aims to publish K8s events for all SparkApplication and SparkCluster state transitions, not only for failure states. The reason of each event is the new state name.

Type	SparkApplication	SparkCluster
Warning	SchedulingFailure, Failed, DriverEvicted, DriverStartTimedOut, DriverReadyTimedOut, ExecutorsStartTimedOut, RunningWithBelowThresholdExecutors, TerminatedWithoutReleaseResources	SchedulingFailure, Failed
Normal	All other states	All other states

An event is published only when the current state changes. Repeated transitions into the same state increment the count of one Event object.

Why are the changes needed?

To make the full lifecycle of Spark resources visible via kubectl describe and kubectl get events.

Does this PR introduce any user-facing change?

No by default. When spark.kubernetes.operator.events.enabled is true, the operator also publishes the new Normal and Warning events above.

How was this patch tested?

Pass the CIs with the newly added unit tests.

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

Generated-by: Claude Opus 5

[dongjoon-hyun]

cc @TQJADE

[dongjoon-hyun]

All tests passed. Could you review this PR when you have some time, please, @peter-toth ?

[peter-toth]

Thanks for the PR, @dongjoon-hyun!

StatusRecorder is the right seam for this, and moving from "current state is a failure" to "every state appended since the previous status" is what makes a single patch carrying two observations report both. I checked the parts that could go wrong and they hold: history keys stay monotonic across the restart trim at ApplicationStatus.java:193-202, so tailMap(prevLastKey + 1) cannot replay; every setLastObservedDriverStatus caller builds a fresh state rather than mutating the current one; and the new equals guard survives the statusCache JSON round trip - I removed it and publishesNoEventWhenTheCurrentStateDidNotChange is the one test that fails. The two items below are both about the classification table being a contract nothing pins.

Non-blocking

• 1. eventTypeOfApplicationStates mirrors the implementation instead of pinning it: it recomputes expected with the same isFailure() || == A || == B expression, so it cannot catch a classification change. I dropped DriverEvicted from ApplicationStateSummary.failures and all 13 tests stayed green, while the docs table would have become wrong. [inline: EventUtilsTest.java:82]

Minor

• 2. The Warning-versus-Normal rule is not written down: InitializedBelowThresholdExecutors being Normal while RunningWithBelowThresholdExecutors is Warning reads as arbitrary until you find the escalation at AppDriverTimeoutObserver.java:78. [inline: EventUtils.java:61]

[peter-toth]

Finding 1. expected is the same expression as EventUtils.eventTypeOf, so this loop asserts that the method agrees with itself. It pins the two NON_FAILURE_WARNING_STATES members, which is worth having, but it delegates the other half of the classification to isFailure() and therefore follows that set wherever it goes.

That matters because the classification is published as a contract: the eight Warning reasons are listed in the PR description and in docs/configuration.md:64-66. I checked what the test would catch by dropping DriverEvicted from ApplicationStateSummary.failures:

$ ./gradlew :spark-operator:test --tests "...EventUtilsTest"
tests="13" skipped="0" failures="0" errors="0"

Green, while DriverEvicted would now publish as Normal and the docs table would be wrong.

An explicit set makes the loop pin the documented table instead:

  @Test
  void eventTypeOfApplicationStates() {
    Set<ApplicationStateSummary> expectedWarnings =
        Set.of(
            ApplicationStateSummary.SchedulingFailure,
            ApplicationStateSummary.Failed,
            ApplicationStateSummary.DriverEvicted,
            ApplicationStateSummary.DriverStartTimedOut,
            ApplicationStateSummary.DriverReadyTimedOut,
            ApplicationStateSummary.ExecutorsStartTimedOut,
            ApplicationStateSummary.RunningWithBelowThresholdExecutors,
            ApplicationStateSummary.TerminatedWithoutReleaseResources);
    for (ApplicationStateSummary summary : ApplicationStateSummary.values()) {
      EventType expected =
          expectedWarnings.contains(summary) ? EventType.WARNING : EventType.NORMAL;
      assertThat(EventUtils.eventTypeOf(summary)).as(summary.name()).isEqualTo(expected);
    }
  }

That also makes the two trailing RunningWithPartialCapacity / InitializedBelowThresholdExecutors assertions redundant, since the loop now covers them by exclusion. The same shape applies to eventTypeOfClusterStates at line 99, where Set.of(SchedulingFailure, Failed) is the whole table.

[dongjoon-hyun]

Thank you, @peter-toth. I updated both tests to pin the documented table with explicit sets, and verified that dropping DriverEvicted from ApplicationStateSummary.failures now fails eventTypeOfApplicationStates.

[peter-toth]

Finding 2. "deserve the attention of users" does not say why InitializedBelowThresholdExecutors is not in this set. Both states mean the app is below its executor threshold, and the startup one is below the minimum required rather than merely under capacity, so on the wording alone it reads like the stronger candidate of the two.

The rule that actually separates them is elsewhere. AppDriverTimeoutObserver.java:78 escalates InitializedBelowThresholdExecutors to ExecutorsStartTimedOut, which is a failure and therefore already a Warning, so the startup case reports itself when it stops being transient. RunningWithBelowThresholdExecutors appears in no timeout branch: it persists silently, so nothing else would ever tell the user. RunningWithPartialCapacity is a third case, documented on the enum as a tolerable capacity level rather than a degradation.

Worth one sentence here, because this is the comment someone will read when they add a state:

  /**
   * States that are not failures but still deserve the attention of users, because nothing else
   * reports them: a state that escalates to a failure on its own, such as
   * {@code InitializedBelowThresholdExecutors} timing out into {@code ExecutorsStartTimedOut},
   * warns through that failure instead and stays normal here.
   */

[dongjoon-hyun]

Thank you. I added the rule to the Javadoc of NON_FAILURE_WARNING_STATES, including why TerminatedWithoutReleaseResources is a warning (it is not meant for production use).

[dongjoon-hyun]

Merged to main