Better handling of task recovery

Author: ssalinas

SingularityService/src/main/java/com/hubspot/singularity/data/TaskManager.java

@@ -1315,15 +1315,29 @@ public void createTaskAndDeletePendingTask(SingularityTask task) {
   public Optional<SingularityTask> tryRepairTask(SingularityTaskId taskId) {
     try {
       Optional<SingularityTask> maybeTask = getTask(taskId); // checks zkCache for task data
-      String path = getTaskPath(taskId);
-      if (maybeTask.isPresent() && !exists(path)) {
-        LOG.info("Found info for task {} from cache not in zk node, rewriting", taskId);
-        save(path, maybeTask.map(taskTranscoder::toBytes));
-        leaderCache.putActiveTask(taskId);
+      if (maybeTask.isPresent() && repairFoundTask(maybeTask.get())) {
+        return maybeTask;
       }
-      return maybeTask;
     } catch (Exception e) {
-      return Optional.empty();
+      LOG.error("Could not find or repair task data for {}", taskId, e);
+    }
+    return Optional.empty();
+  }
+
+  public boolean repairFoundTask(SingularityTask task) {
+    try {
+      String path = getTaskPath(task.getTaskId());
+      LOG.info(
+        "Found info for task {} from cache not in zk node, rewriting",
+        task.getTaskId()
+      );
+      save(path, Optional.of(taskTranscoder.toBytes(task)));
+      leaderCache.putActiveTask(task.getTaskId());
+      taskCache.set(path, task);
+      return true;
+    } catch (Exception e) {
+      LOG.error("Could not repair task data for {}", task.getTaskId(), e);
+      return false;
     }
   }