Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/20449#discussion_r166857525
--- Diff:
core/src/main/scala/org/apache/spark/shuffle/BlockStoreShuffleReader.scala ---
@@ -104,9 +104,18 @@ private[spark] class BlockStoreShuffleReader[K, C](
context.taskMetrics().incMemoryBytesSpilled(sorter.memoryBytesSpilled)
context.taskMetrics().incDiskBytesSpilled(sorter.diskBytesSpilled)
context.taskMetrics().incPeakExecutionMemory(sorter.peakMemoryUsedBytes)
+ // Use completion callback to stop sorter if task was cancelled.
+ context.addTaskCompletionListener(tc => {
+ // Note: we only stop sorter if cancelled as sorter.stop
wouldn't be called in
+ // CompletionIterator. Another way would be making sorter.stop
idempotent.
+ if (tc.isInterrupted()) { sorter.stop() }
--- End diff --
I may be missing something obvious, but seems `ExternalSorter.stop()` is
already idempotent?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
