funrollloops commented on code in PR #58089:
URL: https://github.com/apache/spark/pull/58089#discussion_r3808086960
##########
core/src/main/scala/org/apache/spark/shuffle/streaming/StreamingShuffleWriter.scala:
##########
@@ -257,11 +280,27 @@ class StreamingShuffleWriter[K, V](
val newFuture = future.whenComplete { (client, ex) =>
ex match {
case null => sendToClient(client)
- case _ => buf.release(); done()
+ case error =>
+ buf.release()
+ errorNotifier.markError(error match {
+ case completionException: CompletionException
+ if completionException.getCause != null =>
+ completionException.getCause
+ case _ => error
+ })
+ done()
}
}
// Once the future is completed, stop accumulating CompletionStages.
- client = if (newFuture.isDone) Left(newFuture.join()) else
Right(newFuture)
+ if (newFuture.isDone) {
+ if (newFuture.isCompletedExceptionally) {
+ // Surface the categorized error instead of the
CompletionException from join().
+ throwErrorIfExists()
+ }
+ client = Left(newFuture.join())
Review 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?
##########
core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleWriterSuite.scala:
##########
@@ -229,6 +234,56 @@ class StreamingShuffleWriterSuite
}
}
+ test("writer fails when a reader does not connect before timeout") {
+ val conf = newConf()
+ .set(STREAMING_SHUFFLE_WRITER_CONNECTION_TIMEOUT_MS, 100L)
+ withSpark(new SparkContext("local", "StreamingShuffleWriterSuite", conf))
{ sc =>
+ val context = createTaskContext(sc.conf, 0)
+ try {
+ val writer = newWriter(sc, context)
+ val error = intercept[SparkRuntimeException] {
+ writer.write(Iterator.empty)
+ }
+ checkError(
+ exception = error,
+ condition = "STREAMING_SHUFFLE_WRITER_CONNECTION_TIMEOUT",
+ sqlState = "XXKST",
+ parameters = Map(
+ "shuffleId" -> "0",
+ "writerId" -> "0",
+ "readerId" -> "0",
+ "timeoutMs" -> "100"))
+ } finally {
+ context.markTaskCompleted(None)
+ }
+ }
+ }
+
+ test("send throws the structured error when a reader connection times out") {
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
