Close reader early if there is no more data.

Author: viirya

core/src/main/scala/org/apache/spark/rdd/NewHadoopRDD.scala

@@ -141,6 +141,9 @@ class NewHadoopRDD[K, V](
       override def hasNext: Boolean = {
         if (!finished && !havePair) {
           finished = !reader.nextKeyValue
+          if (finished) {
+            reader.close()
+          }
           havePair = !finished
         }
         !finished
@@ -159,7 +162,9 @@ class NewHadoopRDD[K, V](
 
       private def close() {
         try {
-          reader.close()
+          if (!finished) {
+            reader.close()
+          }
           if (bytesReadCallback.isDefined) {
             inputMetrics.updateBytesRead()
           } else if (split.serializableHadoopSplit.value.isInstanceOf[FileSplit] ||

sql/core/src/main/scala/org/apache/spark/sql/sources/SqlNewHadoopRDD.scala

@@ -161,6 +161,9 @@ private[sql] class SqlNewHadoopRDD[K, V](
       override def hasNext: Boolean = {
         if (!finished && !havePair) {
           finished = !reader.nextKeyValue
+          if (finished) {
+            reader.close()
+          }
           havePair = !finished
         }
         !finished
@@ -179,7 +182,9 @@ private[sql] class SqlNewHadoopRDD[K, V](
 
       private def close() {
         try {
-          reader.close()
+          if (!finished) {
+            reader.close()
+          }
           if (bytesReadCallback.isDefined) {
             inputMetrics.updateBytesRead()
           } else if (split.serializableHadoopSplit.value.isInstanceOf[FileSplit] ||