Mark dagActionsConclude correctly (apache#4010)

Co-authored-by: Urmi Mustafi <umustafi@umustafi-mn2.linkedin.biz>
arjun4084346 · Jul 24, 2024 · dd34ba9 · dd34ba9
1 parent 2ea9622
commit dd34ba9
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 3 deletions.
diff --git a/...ava/org/apache/gobblin/service/modules/orchestration/proc/DeadlineEnforcementDagProc.java b/...ava/org/apache/gobblin/service/modules/orchestration/proc/DeadlineEnforcementDagProc.java
@@ -48,14 +48,14 @@ protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dag
   @Override
   protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag<JobExecutionPlan>> dag,
       DagProcessingEngineMetrics dagProcEngineMetrics) throws IOException {
-    if (validate(dag, dagManagementStateStore)) {
+    if (isDagStillPresent(dag, dagManagementStateStore)) {
       enforceDeadline(dagManagementStateStore, dag.get(), dagProcEngineMetrics);
     } else {
     dagProcEngineMetrics.markDagActionsAct(getDagActionType(), false);
   }
   }
 
-  protected boolean validate(Optional<Dag<JobExecutionPlan>> dag, DagManagementStateStore dagManagementStateStore)
+  protected boolean isDagStillPresent(Optional<Dag<JobExecutionPlan>> dag, DagManagementStateStore dagManagementStateStore)
       throws IOException {
     log.info("Request to enforce deadlines for dag {}", getDagId());
     DagActionStore.DagAction dagAction = getDagTask().getDagAction();

diff --git a/...-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagTask.java b/...-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagTask.java
@@ -63,7 +63,11 @@ public boolean conclude() {
     try {
       this.dagManagementStateStore.deleteDagAction(this.dagAction);
       boolean completedLease = this.leaseObtainedStatus.completeLease();
-      this.dagProcEngineMetrics.markDagActionsConclude(this.dagAction.getDagActionType(), true);
+      if (completedLease) {
+        this.dagProcEngineMetrics.markDagActionsConclude(this.dagAction.getDagActionType(), true);
+      } else {
+        this.dagProcEngineMetrics.markDagActionsConclude(this.dagAction.getDagActionType(), false);
+      }
       return completedLease;
     } catch (IOException e) {
       this.dagProcEngineMetrics.markDagActionsConclude(this.dagAction.getDagActionType(), false);