aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorEvan Jones <ejones@twitter.com>2015-05-08 22:00:39 +0100
committerSean Owen <sowen@cloudera.com>2015-05-08 22:00:39 +0100
commit25889d8d97094325f10fbf52f3b36412f212eeb2 (patch)
treef0fcdb1b58e287d7c2556b64917b0715de138bcd
parent4b3bb0e43ca7e1a27308516608419487b6a844e6 (diff)
downloadspark-25889d8d97094325f10fbf52f3b36412f212eeb2.tar.gz
spark-25889d8d97094325f10fbf52f3b36412f212eeb2.tar.bz2
spark-25889d8d97094325f10fbf52f3b36412f212eeb2.zip
[SPARK-7490] [CORE] [Minor] MapOutputTracker.deserializeMapStatuses: close input streams
GZIPInputStream allocates native memory that is not freed until close() or when the finalizer runs. It is best to close() these streams explicitly. stephenh made the same change for serializeMapStatuses in commit b0d884f0. This is the same change for deserialize. (I ran the unit test suite! it seems to have passed. I did not make a JIRA since this seems "trivial", and the guidelines suggest it is not required for trivial changes) Author: Evan Jones <ejones@twitter.com> Closes #5982 from evanj/master and squashes the following commits: 0d76e85 [Evan Jones] [CORE] MapOutputTracker.deserializeMapStatuses: close input streams
-rw-r--r--core/src/main/scala/org/apache/spark/MapOutputTracker.scala6
1 files changed, 5 insertions, 1 deletions
diff --git a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
index 16072283ed..018422827e 100644
--- a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
+++ b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
@@ -367,7 +367,11 @@ private[spark] object MapOutputTracker extends Logging {
// Opposite of serializeMapStatuses.
def deserializeMapStatuses(bytes: Array[Byte]): Array[MapStatus] = {
val objIn = new ObjectInputStream(new GZIPInputStream(new ByteArrayInputStream(bytes)))
- objIn.readObject().asInstanceOf[Array[MapStatus]]
+ Utils.tryWithSafeFinally {
+ objIn.readObject().asInstanceOf[Array[MapStatus]]
+ } {
+ objIn.close()
+ }
}
// Convert an array of MapStatuses to locations and sizes for a given reduce ID. If