Address more

Author: zsxwing

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

@@ -461,7 +461,8 @@ object HDFSMetadataLog {
         if (minBatchId != startId.get) {
           val missingBatchIds = startId.get to minBatchId
           throw new IllegalStateException(
-            s"batches (${missingBatchIds.mkString(", ")}) don't exist")
+            s"batches (${missingBatchIds.mkString(", ")}) don't exist " +
+              s"(startId: $startId, endId: $endId)")
         }
       }
 
@@ -471,7 +472,8 @@ object HDFSMetadataLog {
         if (maxBatchId != endId.get) {
           val missingBatchIds = maxBatchId to endId.get
           throw new IllegalStateException(
-            s"batches (${missingBatchIds.mkString(", ")}) don't  exist")
+            s"batches (${missingBatchIds.mkString(", ")}) don't  exist " +
+              s"(startId: $startId, endId: $endId)")
         }
       }
     }
@@ -481,7 +483,8 @@ object HDFSMetadataLog {
       val maxBatchId = batchIds.last
       val missingBatchIds = (minBatchId to maxBatchId).toSet -- batchIds
       if (missingBatchIds.nonEmpty) {
-        throw new IllegalStateException(s"batches (${missingBatchIds.mkString(", ")}) don't exist")
+        throw new IllegalStateException(s"batches (${missingBatchIds.mkString(", ")}) " +
+          s"don't exist (startId: $startId, endId: $endId)")
       }
     }
   }

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLogSuite.scala

@@ -264,6 +264,10 @@ class HDFSMetadataLogSuite extends SparkFunSuite with SharedSQLContext {
     import HDFSMetadataLog.verifyBatchIds
     verifyBatchIds(Seq(1L, 2L, 3L), Some(1L), Some(3L))
     verifyBatchIds(Seq(1L), Some(1L), Some(1L))
+    verifyBatchIds(Seq(1L, 2L, 3L), None, Some(3L))
+    verifyBatchIds(Seq(1L, 2L, 3L), Some(1L), None)
+    verifyBatchIds(Seq(1L, 2L, 3L), None, None)
+
     intercept[IllegalStateException](verifyBatchIds(Seq(), Some(1L), None))
     intercept[IllegalStateException](verifyBatchIds(Seq(), None, Some(1L)))
     intercept[IllegalStateException](verifyBatchIds(Seq(), Some(1L), Some(1L)))