[tkaymak]

Fixes #40120. Part of #36841.

SparkStructuredStreamingPipelineResult.getState() now resolves a terminal state from a finished execution without blocking or throwing, CANCELLED when a cancel was requested, otherwise DONE or FAILED. cancel() only requests cancellation while the execution is still running and returns the current state, so it reports RUNNING right after the request and CANCELLED once the execution ended, as discussed on #40103. waitUntilFinish resolves through the same path, so a terminal state is never rewritten.

A cancel after the pipeline already ended reports the real DONE or FAILED and no longer runs the cancel hook. MetricsPusher now stops once the execution ends instead of running until waitUntilFinish is called. The legacy SparkPipelineResult keeps its behavior.

Tests assert CANCELLED behind waitUntilFinish on Spark 3 and 4. testCancelRunsJobCancelHookOnce, testCancelIsAsynchronous, testCancelAfterNormalCompletionReturnsDone and testCancelAfterExceptionalCompletionReturnsFailed fail on master and pass now. Full :runners:spark:4:test and :runners:spark:3:test pass locally. The trigger files run the Spark 4 and SparkStructuredStreaming ValidatesRunner suites.

[github-actions]

📋 Review Summary

This PR refines the state resolution and cancellation behavior in SparkStructuredStreamingPipelineResult so that getState() derives terminal states directly from the underlying execution Future, and cancel() returns the current observed state (RUNNING while shutting down, or the final terminal state once finished). The implementation is clean, avoids prematurely marking still-executing pipelines as CANCELLED or overwriting completed states, and is well-covered by unit and integration tests.

🔍 General Feedback

• Consolidating terminal state transitions inside getState() via Futures.getDone(pipelineExecution) cleanly unifies state resolution across getState(), waitUntilFinish(), and cancel(), while also ensuring background consumers like MetricsPusher observe completion without requiring an explicit waitUntilFinish() call.
• Guarding cancel() with !pipelineExecution.isDone() prevents unnecessary invocation of cancellation hooks and preserves DONE/FAILED states when cancel() is called after pipeline completion.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[tkaymak]

R: @Abacn

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers