Deduplicate code

Author: HeartSaVioR

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala

@@ -162,26 +162,6 @@ abstract class CompactibleFileStreamLog[T <: AnyRef : ClassTag](
     batchAdded
   }
 
-  /**
-   * Return the latest batch Id.
-   *
-   * This method is a complement of getLatest() - while metadata log file per batch tends to be
-   * small, it doesn't apply to the compacted log file. This method only checks for existence of
-   * file to avoid huge cost on reading and deserializing compacted log file.
-   */
-  def getLatestBatchId(): Option[Long] = {
-    val batchIds = fileManager.list(metadataPath, batchFilesFilter)
-      .map(f => pathToBatchId(f.getPath))
-      .sorted(Ordering.Long.reverse)
-    for (batchId <- batchIds) {
-      val batchMetadataFile = batchIdToPath(batchId)
-      if (fileManager.exists(batchMetadataFile)) {
-        return Some(batchId)
-      }
-    }
-    None
-  }
-
   /**
    * CompactibleFileStreamLog maintains logs by itself, and manual purging might break internal
    * state, specifically which latest compaction batch is purged.

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala

@@ -182,19 +182,35 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path:
     }
   }
 
-  override def getLatest(): Option[(Long, T)] = {
+
+  /**
+   * Return the latest batch Id without reading the file. This method only checks for existence of
+   * file to avoid cost on reading and deserializing log file.
+   */
+  def getLatestBatchId(): Option[Long] = {
     val batchIds = fileManager.list(metadataPath, batchFilesFilter)
       .map(f => pathToBatchId(f.getPath))
       .sorted(Ordering.Long.reverse)
     for (batchId <- batchIds) {
-      val batch = get(batchId)
-      if (batch.isDefined) {
-        return Some((batchId, batch.get))
+      val batchMetadataFile = batchIdToPath(batchId)
+      if (fileManager.exists(batchMetadataFile)) {
+        return Some(batchId)
       }
     }
     None
   }
 
+  override def getLatest(): Option[(Long, T)] = {
+    getLatestBatchId().map { batchId =>
+      val content = get(batchId).getOrElse {
+        // This only happens in odd case where the file exists when getLatestBatchId() is called,
+        // but get() doesn't find it.
+        throw new IllegalStateException(s"failed to read log file for batch $batchId")
+      }
+      (batchId, content)
+    }.orElse(None)
+  }
+
   /**
    * Get an array of [FileStatus] referencing batch files.
    * The array is sorted by most recent batch file first to