[GitHub] [hudi] alexeykudinkin commented on a diff in pull request #5141: [HUDI-3724] Fixing closure of ParquetReader

[GitBox <gi...@apache.org>]

alexeykudinkin commented on code in PR #5141:
URL: https://github.com/apache/hudi/pull/5141#discussion_r846670094


##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -333,7 +335,13 @@ object HoodieBaseRelation {
     partitionedFile => {
       val extension = FSUtils.getFileExtension(partitionedFile.filePath)
       if (HoodieFileFormat.PARQUET.getFileExtension.equals(extension)) {
-        parquetReader.apply(partitionedFile)
+        val iter = parquetReader.apply(partitionedFile)
+        if (iter.isInstanceOf[Closeable]) {
+          // register a callback to close parquetReader which will be executed on task completion.
+          // when tasks finished, this method will be called, and release resources.
+          Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ => iter.asInstanceOf[Closeable].close()))

Review Comment:
   @nsivabalan i don't think that was the conclusion -- we concluded that we're okay to use `TaskContext` to close the reader (since Spark 2 isn't properly closing readers itself), but we should unify this logic w/in `HoodieBaseRelation`



-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org