Don't skip job instance if topic doesn't exist (#713)

Author: kevinwallimann

src/main/scala/za/co/absa/hyperdrive/trigger/api/rest/services/HyperdriveOffsetComparisonService.scala

@@ -82,11 +82,8 @@ class HyperdriveOffsetComparisonServiceImpl @Inject() (sparkConfig: SparkConfig,
     val isNewJobInstanceRequiredFut = kafkaEndOffsetsOptFut.flatMap { kafkaEndOffsetsOpt =>
       kafkaBeginningOffsetsOptFut.flatMap { kafkaBeginningOffsetsOpt =>
         (kafkaBeginningOffsetsOpt, kafkaEndOffsetsOpt) match {
-          case (Some(kafkaBeginningOffsets), Some(kafkaEndOffsets)) =>
-            if (kafkaBeginningOffsets.isEmpty) {
-              logger.info(s"Topic ${kafkaParametersOpt.get._1} does not exist. Skipping job instance")
-              Future { false }
-            } else if (offsetsEqual(kafkaBeginningOffsets, kafkaEndOffsets)) {
+          case (Some(kafkaBeginningOffsets), Some(kafkaEndOffsets)) if kafkaBeginningOffsets.nonEmpty =>
+            if (offsetsEqual(kafkaBeginningOffsets, kafkaEndOffsets)) {
               logger.info(s"Topic ${kafkaParametersOpt.get._1} is empty. Skipping job instance")
               Future { false }
             } else {

src/test/scala/za/co/absa/hyperdrive/trigger/api/rest/services/HyperdriveOffsetComparisonServiceTest.scala

@@ -126,7 +126,7 @@ class HyperdriveOffsetComparisonServiceTest extends AsyncFlatSpec with Matchers
     }
   }
 
-  it should "return false if the kafka topic does not exist" in {
+  it should "return true if the kafka topic does not exist" in {
     val config = getSparkConfig
     val underTest = new HyperdriveOffsetComparisonServiceImpl(config.yarn, checkpointService, ugiService, kafkaService)
     val jobParameters = getJobParameters
@@ -138,7 +138,7 @@ class HyperdriveOffsetComparisonServiceTest extends AsyncFlatSpec with Matchers
 
     resultFut.map { result =>
       verify(checkpointService, never()).getLatestOffsetFilePath(any())(any())
-      result shouldBe false
+      result shouldBe true
     }
   }