zeruibao commented on code in PR #58089:
URL: https://github.com/apache/spark/pull/58089#discussion_r3810026987
##########
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:
yaya, let me simulate the real flush trigger route.
--
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]
