Correctly close iterator when reading large log files in Delta Stream…

…ing Source

GitOrigin-RevId: a31e0529bf169d4a1f1e85b9c14982a83a899767

core/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSource.scala

@@ -284,12 +284,16 @@ case class DeltaSource(
     initialState.iterator()
   }
 
-  protected def iteratorLast[T](iter: Iterator[T]): Option[T] = {
-    var last: Option[T] = None
-    while (iter.hasNext) {
-      last = Some(iter.next())
+  protected def iteratorLast[T](iter: ClosableIterator[T]): Option[T] = {
+    try {
+      var last: Option[T] = None
+      while (iter.hasNext) {
+        last = Some(iter.next())
+      }
+      last
+    } finally {
+      iter.close()
     }
-    last
   }
 
   private def getStartingOffset(