DagProcessingThread bug fix (#4109)

* DagProcessingThread bug fix and enhance logging

---------

Co-authored-by: Vaibhav Singhal

Author: vsinghal85

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/DagTask.java

@@ -65,10 +65,15 @@ public boolean conclude() {
       boolean completedLease = this.leaseObtainedStatus.completeLease();
       this.dagProcEngineMetrics.markDagActionsConclude(this.dagAction.getDagActionType(), completedLease);
       return completedLease;
-    } catch (IOException e) {
-      this.dagProcEngineMetrics.markDagActionsConclude(this.dagAction.getDagActionType(), false);
+    } catch (Exception e) {
+      try {
+        this.dagProcEngineMetrics.markDagActionsConclude(this.dagAction.getDagActionType(), false);
+      } catch(Exception ex){
+        log.error("Exception encountered in emitting metrics for dag node ID: {}, dag action type: {}. Stacktrace: ", dagAction.getDagNodeId(), dagAction.getDagActionType(), ex);
+      }
       // TODO: Decide appropriate exception to throw and add to the commit method's signature
-      throw new RuntimeException(e);
+      log.error("Exception encountered in processing this DagTask, for dag node ID: {}, dag action type: {}. Stacktrace: ", dagAction.getDagNodeId(), dagAction.getDagActionType(), e);
+      return false;
     }
   }
 }