Skip to content
Snippets Groups Projects
Commit 25c01c54 authored by jerryshao's avatar jerryshao Committed by Tathagata Das
Browse files

[STREAMING] [MINOR] Close files correctly when iterator is finished in streaming WAL recovery

Currently there's no chance to close the file correctly after the iteration is finished, change to `CompletionIterator` to avoid resource leakage.

Author: jerryshao <saisai.shao@intel.com>

Closes #6050 from jerryshao/close-file-correctly and squashes the following commits:

52dfaf5 [jerryshao] Close files correctly when iterator is finished
parent 8e674331
No related branches found
No related tags found
No related merge requests found
......@@ -26,7 +26,7 @@ import scala.language.postfixOps
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.spark.util.ThreadUtils
import org.apache.spark.util.{CompletionIterator, ThreadUtils}
import org.apache.spark.{Logging, SparkConf}
/**
......@@ -124,7 +124,8 @@ private[streaming] class FileBasedWriteAheadLog(
logFilesToRead.iterator.map { file =>
logDebug(s"Creating log reader with $file")
new FileBasedWriteAheadLogReader(file, hadoopConf)
val reader = new FileBasedWriteAheadLogReader(file, hadoopConf)
CompletionIterator[ByteBuffer, Iterator[ByteBuffer]](reader, reader.close _)
} flatMap { x => x }
}
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment