izchen commented on code in PR #174:
URL: https://github.com/apache/incubator-uniffle/pull/174#discussion_r951161459
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/reader/RssShuffleReader.java:
##########
@@ -201,13 +202,21 @@ public Configuration getHadoopConf() {
RssShuffleDataIterator iterator = new RssShuffleDataIterator<K, C>(
shuffleDependency.serializer(), shuffleReadClient,
readMetrics);
- iterators.add(iterator);
+ CompletionIterator<Product2<K, C>, RssShuffleDataIterator<K, C>>
completionIterator =
Review Comment:
`org.apache.spark.util.CompletionIterator` will clean up as soon as its
wrapped `RssShuffleDataIterator.hasNext` returns `false `:
```
// org.apache.spark.util.CompletionIterator
def hasNext: Boolean = {
val r = iter.hasNext // iter => RssShuffleDataIterator
if (!r && !completed) {
completed = true
// reassign to release resources of highly resource consuming
iterators early
iter = Iterator.empty.asInstanceOf[I]
completion()
}
r
}
def completion(): Unit // completion() => RssShuffleDataIterator.cleanup
```
After this PR, if there is no special case for the `Spark Task`, the timing
of resource cleanup is still when the `RssShuffleDataIterator` ends, not when
the `Spark Task` ends.
This is the same behavior as before the PR.
--
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]
