zeruibao commented on code in PR #58089:
URL: https://github.com/apache/spark/pull/58089#discussion_r3827227334
##########
core/src/main/scala/org/apache/spark/shuffle/streaming/StreamingShuffleWriter.scala:
##########
@@ -257,7 +280,10 @@ 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)
Review Comment:
I removed it. but the timeout test exposed a path where the exceptional
future is never joined. If writing finishes before the connection timeout
(including an empty map partition), close() queues the final message and then
waits for termination acks. No subsequent send calls join(), so without
markError() the writer would waits indefinitely.
##########
core/src/main/scala/org/apache/spark/shuffle/streaming/StreamingShuffleWriter.scala:
##########
@@ -257,7 +280,10 @@ 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)
Review Comment:
just add it back
--
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]
