Repository navigation
Conversation
…ution, cancel() returns the observed state
There was a problem hiding this comment.
📋 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()viaFutures.getDone(pipelineExecution)cleanly unifies state resolution acrossgetState(),waitUntilFinish(), andcancel(), while also ensuring background consumers likeMetricsPusherobserve completion without requiring an explicitwaitUntilFinish()call. - Guarding
cancel()with!pipelineExecution.isDone()prevents unnecessary invocation of cancellation hooks and preservesDONE/FAILEDstates whencancel()is called after pipeline completion.
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
|
R: @Abacn |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
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.waitUntilFinishresolves 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
waitUntilFinishis called. The legacySparkPipelineResultkeeps its behavior.Tests assert CANCELLED behind
waitUntilFinishon Spark 3 and 4.testCancelRunsJobCancelHookOnce,testCancelIsAsynchronous,testCancelAfterNormalCompletionReturnsDoneandtestCancelAfterExceptionalCompletionReturnsFailedfail on master and pass now. Full:runners:spark:4:testand:runners:spark:3:testpass locally. The trigger files run the Spark 4 and SparkStructuredStreaming ValidatesRunner suites.