TEZ-4147. Reduce NN calls in RecoveryService::handleRecoveryEvent

Signed-off-by: Jonathan Eagles <[email protected]>
Raj-Big-data · Apr 17, 2020 · 0495117 · 0495117
1 parent c383ff5
commit 0495117
Showing 1 changed file with 11 additions and 12 deletions.
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/history/recovery/RecoveryService.java b/tez-dag/src/main/java/org/apache/tez/dag/history/recovery/RecoveryService.java
@@ -415,10 +415,10 @@ protected void handleSummaryEvent(TezDAGID dagID,
       if (LOG.isDebugEnabled()) {
         LOG.debug("AppId :" + appContext.getApplicationID() + " summaryPath " + summaryPath);
       }
-      if (!recoveryDirFS.exists(summaryPath)) {
-        summaryStream = recoveryDirFS.create(summaryPath, false,
-            bufferSize);
-      } else {
+      try {
+        summaryStream = recoveryDirFS.create(summaryPath, false, bufferSize);
+      } catch (IOException e) {
+        LOG.error("Error handling summary event, eventType=" + eventType, e);
         createFatalErrorFlagDir();
         return;
       }
@@ -456,16 +456,15 @@ protected void handleRecoveryEvent(DAGHistoryEvent event) throws IOException {
     RecoveryStream recoveryStream = outputStreamMap.get(dagID);
     if (recoveryStream == null) {
       Path dagFilePath = TezCommonUtils.getDAGRecoveryPath(recoveryPath, dagID.toString());
-      if (recoveryDirFS.exists(dagFilePath)) {
-        createFatalErrorFlagDir();
-        return;
-      } else {
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("Opening DAG recovery file in create mode"
-              + ", filePath=" + dagFilePath);
-        }
+
+      try {
         FSDataOutputStream outputStream = recoveryDirFS.create(dagFilePath, false, bufferSize);
+        LOG.debug("Opened DAG recovery file in create mode, filePath={}", dagFilePath);
         recoveryStream = new RecoveryStream(outputStream);
+      } catch (IOException ioe) {
+        LOG.error("Error handling history event, eventType=" + eventType, ioe);
+        createFatalErrorFlagDir();
+        return;
       }
       outputStreamMap.put(dagID, recoveryStream);
     }