#203 Commit (#210)

* #203 Commit

* Committing suggested changes

Author: AshishTiwary

ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/reader/kafka/KafkaStreamReader.scala

@@ -95,8 +95,8 @@ private[reader] class KafkaStreamReader(
   }
 
   private def getStartingOffsets(checkpointLocation: String, configuration: org.apache.hadoop.conf.Configuration): Option[String] = {
-    if (FileUtils.exists(checkpointLocation, configuration)) {
-      Option.empty
+    if (FileUtils.exists(checkpointLocation, configuration) && !FileUtils.isEmpty(checkpointLocation, configuration)) {
+        Option.empty
     }
     else {
       Option(STARTING_OFFSETS_EARLIEST)

ingestor-default/src/test/scala/za/co/absa/hyperdrive/ingestor/implementation/reader/kafka/TestKafkaStreamReader.scala

@@ -25,6 +25,8 @@ import org.scalatest.{BeforeAndAfterEach, FlatSpec}
 import za.co.absa.commons.io.TempDirectory
 import za.co.absa.hyperdrive.ingestor.implementation.reader.kafka.KafkaStreamReaderProps._
 
+import java.nio.file.{Files, Paths}
+
 class TestKafkaStreamReader extends FlatSpec with BeforeAndAfterEach with MockitoSugar {
 
   private val validTopic = "test-topic"
@@ -97,19 +99,43 @@ class TestKafkaStreamReader extends FlatSpec with BeforeAndAfterEach with Mockit
     verify(dataStreamReader, never()).options(validExtraConfs)
   }
 
-  it should "set offsets to earliest if no checkpoint location exists" in {
+  it should "set offsets to earliest if checkpoint location does not exist" in {
     val sparkContext = getMockedSparkContext(stopped = false)
     val dataStreamReader = getMockedDataStreamReader
     val sparkSession = getConfiguredMockedSparkSession(sparkContext, dataStreamReader)
-
     val nonExistent = tempDir.path.resolve("non-existent")
+
     val reader = new KafkaStreamReader(validTopic, validBrokers, nonExistent.toUri.getPath, Map())
     reader.read(sparkSession)
 
     verify(dataStreamReader).option(WORD_STARTING_OFFSETS, STARTING_OFFSETS_EARLIEST)
   }
 
-  it should "set offsets to user-defined property if no checkpoint location exists" in {
+  it should "set offsets to earliest if checkpoint location is empty" in {
+    val sparkContext = getMockedSparkContext(stopped = false)
+    val dataStreamReader = getMockedDataStreamReader
+    val sparkSession = getConfiguredMockedSparkSession(sparkContext, dataStreamReader)
+    Files.createDirectories(Paths.get(s"$tempDirPath/empty1/empty2/empty3"))
+
+    val reader = new KafkaStreamReader(validTopic, validBrokers, tempDirPath, Map())
+    reader.read(sparkSession)
+
+    verify(dataStreamReader).option(WORD_STARTING_OFFSETS, STARTING_OFFSETS_EARLIEST)
+  }
+
+  it should "not set offsets to earliest if a checkpoint location exists and is not empty" in {
+    val sparkContext = getMockedSparkContext(stopped = false)
+    val dataStreamReader = getMockedDataStreamReader
+    val sparkSession = getConfiguredMockedSparkSession(sparkContext, dataStreamReader)
+    Files.createFile(Paths.get(s"$tempDirPath/anyFile"))
+
+    val reader = new KafkaStreamReader(validTopic, validBrokers, tempDirPath, Map())
+    reader.read(sparkSession)
+
+    verify(dataStreamReader, never()).option(WORD_STARTING_OFFSETS, STARTING_OFFSETS_EARLIEST)
+  }
+
+  it should "always set offsets to user-defined property e.g. if checkpoint location does not exist" in {
     val sparkContext = getMockedSparkContext(stopped = false)
     val dataStreamReader = getMockedDataStreamReader
     val sparkSession = getConfiguredMockedSparkSession(sparkContext, dataStreamReader)
@@ -121,15 +147,18 @@ class TestKafkaStreamReader extends FlatSpec with BeforeAndAfterEach with Mockit
     verify(dataStreamReader).options(Map(WORD_STARTING_OFFSETS -> "latest"))
   }
 
-  it should "not set offsets if a checkpoint location exists" in {
+  it should "always set offsets to user-defined property e.g. if checkpoint location exists" in {
     val sparkContext = getMockedSparkContext(stopped = false)
     val dataStreamReader = getMockedDataStreamReader
     val sparkSession = getConfiguredMockedSparkSession(sparkContext, dataStreamReader)
 
-    val reader = new KafkaStreamReader(validTopic, validBrokers, tempDirPath, Map())
+    Files.createFile(Paths.get(s"$tempDirPath/anyFile"))
+
+    val nonExistent = tempDir.path.resolve("non-existent")
+    val reader = new KafkaStreamReader(validTopic, validBrokers, nonExistent.toUri.getPath, Map(WORD_STARTING_OFFSETS -> "latest"))
     reader.read(sparkSession)
 
-    verify(dataStreamReader, never()).option(WORD_STARTING_OFFSETS, STARTING_OFFSETS_EARLIEST)
+    verify(dataStreamReader).options(Map(WORD_STARTING_OFFSETS -> "latest"))
   }
 
   private def getMockedSparkContext(stopped: Boolean): SparkContext = {

shared/src/main/scala/za/co/absa/hyperdrive/shared/utils/FileUtils.scala

@@ -37,7 +37,7 @@ private[hyperdrive] object FileUtils {
   def isEmpty(directory: String, configuration: Configuration): Boolean = {
     val fs = getFileSystem(configuration)
     val path = new Path(directory)
-    fs.exists(path) && !fs.listLocatedStatus(path).hasNext
+    fs.exists(path) &&  !fs.listFiles(path, true).hasNext
   }
 
   private def getFileSystem(configuration: Configuration): FileSystem = {

shared/src/test/scala/za/co/absa/hyperdrive/shared/utils/TestFileUtils.scala

@@ -16,8 +16,7 @@
 package za.co.absa.hyperdrive.shared.utils
 
 import java.util.UUID
-
-import org.apache.hadoop.fs.{FileSystem, Path}
+import org.apache.hadoop.fs.{FileSystem, LocatedFileStatus, Path, RemoteIterator}
 import org.scalatest.{BeforeAndAfter, FlatSpec, Matchers}
 import za.co.absa.commons.io.TempDirectory
 import za.co.absa.commons.spark.SparkTestBase
@@ -49,11 +48,22 @@ class TestFileUtils extends FlatSpec with Matchers with SparkTestBase with Befor
     FileUtils.isEmpty(directory, config) shouldBe true
   }
 
+  it should "return true if the directory only contains other directories, but no files" in {
+    // given
+    val directory = s"$baseDirPath/empty1"
+    val subDirectory = s"$baseDirPath/empty1/empty2/empty3"
+    fs.mkdirs(new Path(subDirectory))
+
+    // when, then
+    FileUtils.isEmpty(directory, config) shouldBe true
+  }
+
   it should "return false if the directory is not empty" in {
     // given
     val directory = s"$baseDirPath/empty"
+    val subDirectory = s"$baseDirPath/empty/empty2"
     fs.mkdirs(new Path(directory))
-    fs.create(new Path(directory, "_INFO"))
+    fs.create(new Path(subDirectory, "_INFO"))
 
     // when, then
     FileUtils.isEmpty(directory, config) shouldBe false