[SPARK-58848][SS] Prevent unbounded shuffle reader connection waits - #58089
[SPARK-58848][SS] Prevent unbounded shuffle reader connection waits#58089zeruibao wants to merge 2 commits into
Conversation
funrollloops
left a comment
funrollloops
left a comment
There was a problem hiding this comment.
Thanks for adding this missing timeout.
| if (newFuture.isCompletedExceptionally) { | ||
| // Surface the categorized error instead of the CompletionException from join(). | ||
| throwErrorIfExists() | ||
| } | ||
| client = Left(newFuture.join()) |
There was a problem hiding this comment.
Why is this inconsistent with lines 287 above? And why do we need both this and line 287?
Could we instead consistently unwrap the completion exceptions, if that's really what we want, at the existing exception handling block in write() and in the flush thread?
There was a problem hiding this comment.
yeah, I agree it would be nicer.
| } | ||
| } | ||
|
|
||
| test("send throws the structured error when a reader connection times out") { |
There was a problem hiding this comment.
can we instead write 1 row, sleep > the timeout and flush interval, and write a second row? This should test exceptions being inside the flush thread.
As written this test doesn't really cover anything different than the test case above. It also depends on internal implementation details and so is likely to need changes if the code changes.
There was a problem hiding this comment.
yaya, let me simulate the real flush trigger route.
|
Thank you @zeruibao! Please add a component tag to the PR title. |
|
Also, thank you @jerrypeng and @funrollloops for review! |
| .createWithDefault(50) | ||
|
|
||
| private[spark] val STREAMING_SHUFFLE_WRITER_CONNECTION_TIMEOUT_MS = | ||
| ConfigBuilder("spark.shuffle.streaming.writerConnectionTimeoutMs") |
There was a problem hiding this comment.
it's a timeConf which accepts a human-readable string literal, unnecessary to end with "Ms"
zeruibao
changed the title
Unwrap asynchronous failures at thread boundaries and make the flush-thread timeout test deterministic while adopting a duration-style config name.
5 participants
What changes were proposed in this pull request?
Prevent unbounded shuffle reader connection waits. Instead, timeout after 1 hour.
Why are the changes needed?
unbounded shuffle reader connection waits is unsafe.
Does this PR introduce any user-facing change?
No
How was this patch tested?
UT
Was this patch authored or co-authored using generative AI tooling?
Cursor