jerqi commented on code in PR #174:
URL: https://github.com/apache/incubator-uniffle/pull/174#discussion_r951170368
##########
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.
OK, I got it. Good catch.
--
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]
