Feature/395 prevent self deactivation (#397)

* Use the same heartbeat for updateHeartbeat and deactivation

* Never deactivate own instance

* Additional log output

Author: kevinwallimann

src/main/scala/za/co/absa/hyperdrive/trigger/persistance/SchedulerInstanceRepository.scala

@@ -28,9 +28,9 @@ import scala.util.{Failure, Success}
 trait SchedulerInstanceRepository extends Repository {
   def insertInstance()(implicit ec: ExecutionContext): Future[Long]
 
-  def updateHeartbeat(id: Long)(implicit ec: ExecutionContext): Future[Int]
+  def updateHeartbeat(id: Long, newHeartbeat: LocalDateTime)(implicit ec: ExecutionContext): Future[Int]
 
-  def deactivateLaggingInstances(currentHeartbeat: LocalDateTime, lagTolerance: Duration)(implicit ec: ExecutionContext): Future[Int]
+  def deactivateLaggingInstances(instanceId: Long, currentHeartbeat: LocalDateTime, lagTolerance: Duration)(implicit ec: ExecutionContext): Future[Int]
 
   def getAllInstances()(implicit ec: ExecutionContext): Future[Seq[SchedulerInstance]]
 }
@@ -55,16 +55,17 @@ class SchedulerInstanceRepositoryImpl extends SchedulerInstanceRepository {
     }
   }
 
-  override def updateHeartbeat(id: Long)(implicit ec: ExecutionContext): Future[Int] = db.run {
+  override def updateHeartbeat(id: Long, newHeartbeat: LocalDateTime)(implicit ec: ExecutionContext): Future[Int] = db.run {
     schedulerInstanceTable.filter(_.id === id)
       .filter(_.status === LiteralColumn[SchedulerInstanceStatus](SchedulerInstanceStatuses.Active))
       .map(_.lastHeartbeat)
-      .update(LocalDateTime.now())
+      .update(newHeartbeat)
   }
 
-  override def deactivateLaggingInstances(currentHeartbeat: LocalDateTime, lagTolerance: Duration)(implicit ec: ExecutionContext): Future[Int] = db.run {
+  override def deactivateLaggingInstances(instanceId: Long, currentHeartbeat: LocalDateTime, lagTolerance: Duration)(implicit ec: ExecutionContext): Future[Int] = db.run {
     schedulerInstanceTable.filter(i => i.lastHeartbeat < currentHeartbeat.minusSeconds(lagTolerance.getSeconds))
       .filter(_.status === LiteralColumn[SchedulerInstanceStatus](SchedulerInstanceStatuses.Active))
+      .filter(_.id =!= instanceId)
       .map(_.status)
       .update(SchedulerInstanceStatuses.Deactivated)
   }

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/JobScheduler.scala

@@ -91,6 +91,7 @@ class JobScheduler @Inject()(sensors: Sensors, executors: Executors, dagInstance
       runningAssignWorkflows = workflowBalancer.getAssignedWorkflows(runningDags.keys.map(_.workflowId).toSeq)
         .recover {
           case e: SchedulerInstanceAlreadyDeactivatedException =>
+            logger.error("Stopping scheduler because the instance has already been deactivated", e)
             stopManager()
             throw e
         }

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/cluster/SchedulerInstanceService.scala

@@ -15,8 +15,9 @@
 
 package za.co.absa.hyperdrive.trigger.scheduler.cluster
 
-import java.time.{Duration, LocalDateTime}
+import org.slf4j.LoggerFactory
 
+import java.time.{Duration, LocalDateTime}
 import javax.inject.Inject
 import org.springframework.stereotype.Service
 import za.co.absa.hyperdrive.trigger.models.SchedulerInstance
@@ -33,18 +34,21 @@ trait SchedulerInstanceService {
 
 @Service
 class SchedulerInstanceServiceImpl @Inject()(schedulerInstanceRepository: SchedulerInstanceRepository) extends SchedulerInstanceService {
+  private val logger = LoggerFactory.getLogger(this.getClass)
 
   override def registerNewInstance()(implicit ec: ExecutionContext): Future[Long] = schedulerInstanceRepository.insertInstance()
 
   override def updateSchedulerStatus(instanceId: Long, lagThreshold: Duration)(implicit ec: ExecutionContext): Future[Seq[SchedulerInstance]] = {
+    val currentHeartbeat = LocalDateTime.now()
     for {
-      updatedCount <- schedulerInstanceRepository.updateHeartbeat(instanceId)
+      updatedCount <- schedulerInstanceRepository.updateHeartbeat(instanceId, currentHeartbeat)
       _ <- if (updatedCount == 0) {
         Future.failed(new SchedulerInstanceAlreadyDeactivatedException)
       } else {
         Future{}
       }
-      _ <- schedulerInstanceRepository.deactivateLaggingInstances(LocalDateTime.now(), lagThreshold)
+      deactivatedCount <- schedulerInstanceRepository.deactivateLaggingInstances(instanceId, currentHeartbeat, lagThreshold)
+      _ = if (deactivatedCount != 0) logger.debug(s"Deactivated $deactivatedCount instances at current heartbeat $currentHeartbeat")
       allInstances <- schedulerInstanceRepository.getAllInstances()
     } yield allInstances
   }

src/test/scala/za/co/absa/hyperdrive/trigger/persistance/SchedulerInstanceRepositoryTest.scala

@@ -61,30 +61,31 @@ class SchedulerInstanceRepositoryTest extends FlatSpec with Matchers with Before
   }
 
   "updateHeartbeat" should "update the last heartbeat of an active instance" in {
-    val nowMinusOne = LocalDateTime.now().minusNanos(1L)
-    val result = await(schedulerInstanceRepository.updateHeartbeat(11))
+    val newHeartbeat = LocalDateTime.now()
+    val result = await(schedulerInstanceRepository.updateHeartbeat(11, newHeartbeat))
     val updatedInstance = await(db.run(schedulerInstanceTable.filter(_.id === 11L).result.head))
 
     result shouldBe 1
-    updatedInstance.lastHeartbeat.isAfter(nowMinusOne) shouldBe true
+    updatedInstance.lastHeartbeat shouldBe newHeartbeat
   }
 
   it should "not update a deactivated instance" in {
-    val result = await(schedulerInstanceRepository.updateHeartbeat(31L))
+    val newHeartbeat = LocalDateTime.now()
+    val result = await(schedulerInstanceRepository.updateHeartbeat(31L, newHeartbeat))
     result shouldBe 0
   }
 
-  "deactivateLaggingInstances" should "deactivate lagging instances" in {
+  "deactivateLaggingInstances" should "deactivate lagging instances, except own instance" in {
     val localTime = LocalDateTime.of(2020, 1, 1, 2, 30, 28)
     val lagTolerance = Duration.ofSeconds(20L)
 
-    val result = await(schedulerInstanceRepository.deactivateLaggingInstances(localTime, lagTolerance))
+    val result = await(schedulerInstanceRepository.deactivateLaggingInstances(21L, localTime, lagTolerance))
     val allInstances = await(db.run(schedulerInstanceTable.result))
 
-    result shouldBe 2
+    result shouldBe 1
     allInstances
       .filter(_.status == SchedulerInstanceStatuses.Deactivated)
-      .map(_.id) should contain theSameElementsAs Seq(21L, 22L, 31L)
+      .map(_.id) should contain theSameElementsAs Seq(22L, 31L)
   }
 
   "getAllInstances" should "return all instances" in {

src/test/scala/za/co/absa/hyperdrive/trigger/scheduler/cluster/SchedulerInstanceServiceTest.scala

@@ -61,30 +61,30 @@ class SchedulerInstanceServiceTest extends AsyncFlatSpec with MockitoSugar with
       SchedulerInstance(23, SchedulerInstanceStatuses.Active, LocalDateTime.now()),
       SchedulerInstance(24, SchedulerInstanceStatuses.Active, LocalDateTime.now())
     )
-    when(schedulerInstanceRepository.updateHeartbeat(any())(any[ExecutionContext])).thenReturn(Future{1})
+    when(schedulerInstanceRepository.updateHeartbeat(any(), any())(any[ExecutionContext])).thenReturn(Future{1})
     when(schedulerInstanceRepository.getAllInstances()(any[ExecutionContext])).thenReturn(Future{instances})
-    when(schedulerInstanceRepository.deactivateLaggingInstances(any(), any())(any[ExecutionContext])).thenReturn(Future{0})
+    when(schedulerInstanceRepository.deactivateLaggingInstances(any(), any(), any())(any[ExecutionContext])).thenReturn(Future{0})
 
     // when
     val result = await(underTest.updateSchedulerStatus(23L, lagThreshold))
 
     // then
     result shouldBe instances
-    verify(schedulerInstanceRepository, times(1)).updateHeartbeat(eqTo(23L))(any())
-    verify(schedulerInstanceRepository, times(1)).deactivateLaggingInstances(any(), eqTo(lagThreshold))(any())
+    verify(schedulerInstanceRepository, times(1)).updateHeartbeat(eqTo(23L), any())(any())
+    verify(schedulerInstanceRepository, times(1)).deactivateLaggingInstances(eqTo(23L), any(), eqTo(lagThreshold))(any())
     succeed
   }
 
   it should "throw an exception if the heartbeat could not be updated" in {
     // given
     val lagThreshold = Duration.ofSeconds(5L)
-    when(schedulerInstanceRepository.updateHeartbeat(any())(any[ExecutionContext])).thenReturn(Future{0})
+    when(schedulerInstanceRepository.updateHeartbeat(any(), any())(any[ExecutionContext])).thenReturn(Future{0})
 
     // when
     the [SchedulerInstanceAlreadyDeactivatedException] thrownBy await(underTest.updateSchedulerStatus(23L, lagThreshold))
 
     // then
-    verify(schedulerInstanceRepository, never).deactivateLaggingInstances(any(), any())(any())
+    verify(schedulerInstanceRepository, never).deactivateLaggingInstances(any(), any(), any())(any())
     verify(schedulerInstanceRepository, never).getAllInstances()(any())
     succeed
   }