[PR] add resume dag proc [gobblin]

["arjun4084346 (via GitHub)" <gi...@apache.org>]

arjun4084346 opened a new pull request, #3938:
URL: https://github.com/apache/gobblin/pull/3938

   Dear Gobblin maintainers,
   
   Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
   
   
   ### JIRA
   - [ ] My PR addresses the following [Gobblin JIRA](https://issues.apache.org/jira/browse/GOBBLIN/) issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
       - https://issues.apache.org/jira/browse/GOBBLIN-XXX
   
   
   ### Description
   - [ ] Here are some details about my PR, including screenshots (if applicable):
   
   
   ### Tests
   - [ ] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
   
   
   ### Commits
   - [ ] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
       1. Subject is separated from body by a blank line
       2. Subject is limited to 50 characters
       3. Subject does not end with a period
       4. Subject uses the imperative mood ("add", not "adding")
       5. Body wraps at 72 characters
       6. Body explains "what" and "why", not "how"
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["phet (via GitHub)" <gi...@apache.org>]

phet commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1596100187


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -82,12 +82,14 @@ protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag
         Map<String, String> jobMetadata = TimingEventUtils.getJobMetadata(Maps.newHashMap(), node.getValue());
         eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_PENDING_RESUME).stop(jobMetadata);
       }
-      // Set flowStartTime so that flow SLA will be based on current time instead of original flow
+      // Set flowStartTime so that flow start deadline and flow completion deadline will be based on current time instead of original flow
       node.getValue().setFlowStartTime(flowResumeTime);
     }
 
+    // these two statements effectively move the dag from failed dag store to (running) dag store

Review Comment:
   check for whether we do find a failed dag, but the same one is also in the dag state store (so presumably a prior exec had failed between line 90 and 91)



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["phet (via GitHub)" <gi...@apache.org>]

phet commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1594899119


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that resumes a dag and submits the job that failed/killed previously.
+ */
+@Slf4j
+public class ResumeDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>> {
+
+  public ResumeDagProc(ResumeDagTask resumeDagTask) {
+    super(resumeDagTask);
+  }
+
+  @Override
+  protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore)
+      throws IOException {
+   return dagManagementStateStore.getFailedDag(getDagId());
+  }
+
+  @Override
+  protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag<JobExecutionPlan>> dag)
+      throws IOException {
+    log.info("Request to resume dag {}", getDagId());
+
+    if (!dag.isPresent()) {
+      // todo - add a metric here
+      log.error("Dag " + dagId + " was not found in dag state store");
+      return;
+    }
+
+    long flowResumeTime = System.currentTimeMillis();
+
+    // Set the flow and its failed or cancelled nodes to PENDING_RESUME so that the flow will be resumed from the point before it failed
+    DagManagerUtils.emitFlowEvent(eventSubmitter, dag.get(), TimingEvent.FlowTimings.FLOW_PENDING_RESUME);
+
+    for (Dag.DagNode<JobExecutionPlan> node : dag.get().getNodes()) {
+      ExecutionStatus executionStatus = node.getValue().getExecutionStatus();
+      if (executionStatus.equals(FAILED) || executionStatus.equals(CANCELLED)) {
+        node.getValue().setExecutionStatus(PENDING_RESUME);
+        // reset currentAttempts because we do not want to count previous execution's attempts in deciding whether to retry a job
+        node.getValue().setCurrentAttempts(0);
+        DagManagerUtils.incrementJobGeneration(node);

Review Comment:
   glad to see the comment, and would be better if contextualizing the currentAttempts "reset" when mentioning the "generation" increment



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["phet (via GitHub)" <gi...@apache.org>]

phet commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1597013555


##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProcTest.java:
##########
@@ -100,9 +102,17 @@ public void resumeDag()
     resumeDagProc.process(this.dagManagementStateStore);
 
     SpecProducer<Spec> specProducer = DagManagerUtils.getSpecProducer(dag.getNodes().get(1));
+    List<SpecProducer<Spec>> otherSpecProducers = dag.getNodes().stream().map(node -> {
+      try {
+        return DagManagerUtils.getSpecProducer(node);
+      } catch (ExecutionException | InterruptedException e) {
+        throw new RuntimeException(e);
+      }
+    }).filter(sp -> specProducer != sp).collect(Collectors.toList());
     int expectedNumOfResumedJobs = 1; // = number of resumed nodes
 
     Mockito.verify(specProducer, Mockito.times(expectedNumOfResumedJobs)).addSpec(any());
     Mockito.verify(this.dagManagementStateStore, Mockito.times(expectedNumOfResumedJobs)).addDagNodeState(any(), any());
+    otherSpecProducers.forEach(sp -> Mockito.verify(sp, Mockito.times(0)).addSpec(any()));

Review Comment:
   nit: `Mockito.never()`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["phet (via GitHub)" <gi...@apache.org>]

phet commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1596009979


##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagManagementDagActionStoreChangeMonitor.java:
##########
@@ -63,10 +63,11 @@ protected void handleDagAction(DagActionStore.DagAction dagAction, boolean isSta
         case LAUNCH :
         case REEVALUATE :
         case KILL :
+        case RESUME:

Review Comment:
   makes sense... although IME that may not be immediately apparent to future maintainers



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["umustafi (via GitHub)" <gi...@apache.org>]

umustafi commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1583986914


##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProcTest.java:
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.api.FlowSpec;
+import org.apache.gobblin.runtime.api.Spec;
+import org.apache.gobblin.runtime.api.SpecProducer;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagActionStore;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.DagManagerTest;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.MostlyMySqlDagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.MostlyMySqlDagManagementStateStoreTest;
+import org.apache.gobblin.service.modules.orchestration.MysqlDagActionStore;
+import org.apache.gobblin.service.modules.orchestration.task.KillDagTask;
+import org.apache.gobblin.service.modules.orchestration.task.LaunchDagTask;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.service.modules.utils.FlowCompilationValidationHelper;
+import org.apache.gobblin.service.monitoring.JobStatus;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+
+
+public class ResumeDagProcTest {
+  private MostlyMySqlDagManagementStateStore dagManagementStateStore;
+
+  @BeforeClass
+  public void setUp() throws Exception {
+    this.dagManagementStateStore = spy(MostlyMySqlDagManagementStateStoreTest.getDummyDMSS(TestMetastoreDatabaseFactory.get()));
+    doReturn(FlowSpec.builder().build()).when(this.dagManagementStateStore).getFlowSpec(any());
+    doNothing().when(this.dagManagementStateStore).tryAcquireQuota(any());
+    doNothing().when(this.dagManagementStateStore).addDagNodeState(any(), any());
+  }
+
+  @Test
+  public void killDag() throws IOException, URISyntaxException {

Review Comment:
   add doc string to summarize what this test does, same for `killDagNode`. These long tests take a while to understand. 



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that kills all the nodes of a dag.
+ * If the dag action has job name set, then it kills only that particular job/dagNode.
+ */
+@Slf4j
+public class ResumeDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>> {
+
+  public ResumeDagProc(ResumeDagTask resumeDagTask) {
+    super(resumeDagTask);
+  }
+
+  @Override
+  protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore)
+      throws IOException {
+   return dagManagementStateStore.getFailedDag(getDagId());
+  }
+
+  @Override
+  protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag<JobExecutionPlan>> dag)
+      throws IOException {
+    log.info("Request to resume dag {}", getDagId());
+
+    if (!dag.isPresent()) {
+      // todo - add a metric here
+      log.error("Dag " + dagId + " was not found in dag state store");
+      return;
+    }
+
+    long flowResumeTime = System.currentTimeMillis();
+
+    // Set the flow and its failed or cancelled nodes to PENDING_RESUME so that the flow will be resumed from the point before it failed
+    DagManagerUtils.emitFlowEvent(eventSubmitter, dag.get(), TimingEvent.FlowTimings.FLOW_PENDING_RESUME);
+
+    for (Dag.DagNode<JobExecutionPlan> node : dag.get().getNodes()) {
+      ExecutionStatus executionStatus = node.getValue().getExecutionStatus();
+      if (executionStatus.equals(FAILED) || executionStatus.equals(CANCELLED)) {
+        node.getValue().setExecutionStatus(PENDING_RESUME);
+        // reset currentAttempts because we do not want to count previous execution's attempts in deciding whether to retry a job
+        node.getValue().setCurrentAttempts(0);
+        DagManagerUtils.incrementJobGeneration(node);
+        Map<String, String> jobMetadata = TimingEventUtils.getJobMetadata(Maps.newHashMap(), node.getValue());
+        eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_PENDING_RESUME).stop(jobMetadata);
+      }
+      // Set flowStartTime so that flow SLA will be based on current time instead of original flow
+      node.getValue().setFlowStartTime(flowResumeTime);

Review Comment:
   is most of this code unchanged from DagManager? if so can we put in a common func (defined in this class) to be re-used across both places? 



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that kills all the nodes of a dag.

Review Comment:
   this should be resume not kill (copy paste error it seems)



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["arjun4084346 (via GitHub)" <gi...@apache.org>]

arjun4084346 commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1593288411


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that kills all the nodes of a dag.
+ * If the dag action has job name set, then it kills only that particular job/dagNode.
+ */
+@Slf4j
+public class ResumeDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>> {
+
+  public ResumeDagProc(ResumeDagTask resumeDagTask) {
+    super(resumeDagTask);
+  }
+
+  @Override
+  protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore)
+      throws IOException {
+   return dagManagementStateStore.getFailedDag(getDagId());
+  }
+
+  @Override
+  protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag<JobExecutionPlan>> dag)
+      throws IOException {
+    log.info("Request to resume dag {}", getDagId());
+
+    if (!dag.isPresent()) {
+      // todo - add a metric here
+      log.error("Dag " + dagId + " was not found in dag state store");
+      return;
+    }
+
+    long flowResumeTime = System.currentTimeMillis();
+
+    // Set the flow and its failed or cancelled nodes to PENDING_RESUME so that the flow will be resumed from the point before it failed
+    DagManagerUtils.emitFlowEvent(eventSubmitter, dag.get(), TimingEvent.FlowTimings.FLOW_PENDING_RESUME);
+
+    for (Dag.DagNode<JobExecutionPlan> node : dag.get().getNodes()) {
+      ExecutionStatus executionStatus = node.getValue().getExecutionStatus();
+      if (executionStatus.equals(FAILED) || executionStatus.equals(CANCELLED)) {
+        node.getValue().setExecutionStatus(PENDING_RESUME);
+        // reset currentAttempts because we do not want to count previous execution's attempts in deciding whether to retry a job
+        node.getValue().setCurrentAttempts(0);
+        DagManagerUtils.incrementJobGeneration(node);
+        Map<String, String> jobMetadata = TimingEventUtils.getJobMetadata(Maps.newHashMap(), node.getValue());
+        eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_PENDING_RESUME).stop(jobMetadata);
+      }
+      // Set flowStartTime so that flow SLA will be based on current time instead of original flow
+      node.getValue().setFlowStartTime(flowResumeTime);

Review Comment:
   there might be some very small changes here and there that makes it less useful to create a common func. 
   and then DagManager will anyway be removed eventually.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["phet (via GitHub)" <gi...@apache.org>]

phet commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1596217894


##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProcTest.java:
##########
@@ -101,24 +99,10 @@ public void resumeDag() throws IOException, URISyntaxException {
         null, mock(DagActionStore.class)));
     resumeDagProc.process(this.dagManagementStateStore);
 
-    List<SpecProducer<Spec>> specProducers = dag.getNodes().stream().map(n -> {
-      try {
-        return DagManagerUtils.getSpecProducer(n);
-      } catch (ExecutionException | InterruptedException e) {
-        throw new RuntimeException(e);
-      }
-    }).collect(Collectors.toList());
+    SpecProducer<Spec> specProducer = DagManagerUtils.getSpecProducer(dag.getNodes().get(1));
     int expectedNumOfResumedJobs = 1; // = number of resumed nodes
 
-    long resumedJobCount = specProducers.stream()
-        .mapToLong(p -> Mockito.mockingDetails(p)
-            .getInvocations()
-            .stream()
-            .filter(a -> a.getMethod().getName().equals("addSpec"))
-            .count())
-        .sum();
-
-    Assert.assertEquals(resumedJobCount, expectedNumOfResumedJobs);
+    Mockito.verify(specProducer, Mockito.times(expectedNumOfResumedJobs)).addSpec(any());

Review Comment:
   excellent!
   
   ...and for the other `SpecProducer`s, don't we want:
   ```
   Mockito.verifyZeroInteractions(anotherSP);
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["phet (via GitHub)" <gi...@apache.org>]

phet commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1594897042


##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagManagementDagActionStoreChangeMonitor.java:
##########
@@ -63,10 +63,11 @@ protected void handleDagAction(DagActionStore.DagAction dagAction, boolean isSta
         case LAUNCH :
         case REEVALUATE :
         case KILL :
+        case RESUME:

Review Comment:
   nit: let's alphabetize these



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/task/ResumeDagTask.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.task;
+
+import org.apache.gobblin.service.modules.orchestration.DagActionStore;
+import org.apache.gobblin.service.modules.orchestration.DagTaskVisitor;
+import org.apache.gobblin.service.modules.orchestration.LeaseAttemptStatus;
+
+
+/**
+ * A {@link DagTask} responsible for killing running jobs.

Review Comment:
   needs correction
   
   also, I don't believe we put a blank line between javadoc and the class



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that resumes a dag and submits the job that failed/killed previously.
+ */
+@Slf4j
+public class ResumeDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>> {
+
+  public ResumeDagProc(ResumeDagTask resumeDagTask) {
+    super(resumeDagTask);
+  }
+
+  @Override
+  protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore)
+      throws IOException {
+   return dagManagementStateStore.getFailedDag(getDagId());
+  }
+
+  @Override
+  protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag<JobExecutionPlan>> dag)
+      throws IOException {
+    log.info("Request to resume dag {}", getDagId());
+
+    if (!dag.isPresent()) {
+      // todo - add a metric here
+      log.error("Dag " + dagId + " was not found in dag state store");
+      return;
+    }
+
+    long flowResumeTime = System.currentTimeMillis();
+
+    // Set the flow and its failed or cancelled nodes to PENDING_RESUME so that the flow will be resumed from the point before it failed
+    DagManagerUtils.emitFlowEvent(eventSubmitter, dag.get(), TimingEvent.FlowTimings.FLOW_PENDING_RESUME);
+
+    for (Dag.DagNode<JobExecutionPlan> node : dag.get().getNodes()) {
+      ExecutionStatus executionStatus = node.getValue().getExecutionStatus();
+      if (executionStatus.equals(FAILED) || executionStatus.equals(CANCELLED)) {
+        node.getValue().setExecutionStatus(PENDING_RESUME);
+        // reset currentAttempts because we do not want to count previous execution's attempts in deciding whether to retry a job
+        node.getValue().setCurrentAttempts(0);
+        DagManagerUtils.incrementJobGeneration(node);
+        Map<String, String> jobMetadata = TimingEventUtils.getJobMetadata(Maps.newHashMap(), node.getValue());
+        eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_PENDING_RESUME).stop(jobMetadata);
+      }
+      // Set flowStartTime so that flow SLA will be based on current time instead of original flow
+      node.getValue().setFlowStartTime(flowResumeTime);
+    }
+
+    dagManagementStateStore.checkpointDag(dag.get());
+    dagManagementStateStore.deleteFailedDag(dag.get());
+    resumeDag(dagManagementStateStore, dag.get());
+  }
+
+  private void resumeDag(DagManagementStateStore dagManagementStateStore, Dag<JobExecutionPlan> dag) {
+    Set<Dag.DagNode<JobExecutionPlan>> nextNodes = DagManagerUtils.getNext(dag);
+
+    if (nextNodes.size() > 1) {
+      handleMultipleJobs(nextNodes);
+    }
+
+    //Submit jobs from the dag ready for execution.
+    for (Dag.DagNode<JobExecutionPlan> dagNode : nextNodes) {
+      DagProcUtils.submitJobToExecutor(dagManagementStateStore, dagNode, getDagId());
+      log.info("Submitted job {} for dagId {}", DagManagerUtils.getJobName(dagNode), getDagId());
+    }
+  }
+
+  private void handleMultipleJobs(Set<Dag.DagNode<JobExecutionPlan>> nextNodes) {
+    throw new UnsupportedOperationException("More than one start job is not allowed");
+  }

Review Comment:
   after this PR, I believe we'll be ready to implement this (all three cases)



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that resumes a dag and submits the job that failed/killed previously.
+ */
+@Slf4j
+public class ResumeDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>> {
+
+  public ResumeDagProc(ResumeDagTask resumeDagTask) {
+    super(resumeDagTask);
+  }
+
+  @Override
+  protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore)
+      throws IOException {
+   return dagManagementStateStore.getFailedDag(getDagId());
+  }
+
+  @Override
+  protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag<JobExecutionPlan>> dag)
+      throws IOException {
+    log.info("Request to resume dag {}", getDagId());
+
+    if (!dag.isPresent()) {
+      // todo - add a metric here
+      log.error("Dag " + dagId + " was not found in dag state store");
+      return;
+    }
+
+    long flowResumeTime = System.currentTimeMillis();
+
+    // Set the flow and its failed or cancelled nodes to PENDING_RESUME so that the flow will be resumed from the point before it failed
+    DagManagerUtils.emitFlowEvent(eventSubmitter, dag.get(), TimingEvent.FlowTimings.FLOW_PENDING_RESUME);
+
+    for (Dag.DagNode<JobExecutionPlan> node : dag.get().getNodes()) {
+      ExecutionStatus executionStatus = node.getValue().getExecutionStatus();
+      if (executionStatus.equals(FAILED) || executionStatus.equals(CANCELLED)) {
+        node.getValue().setExecutionStatus(PENDING_RESUME);
+        // reset currentAttempts because we do not want to count previous execution's attempts in deciding whether to retry a job
+        node.getValue().setCurrentAttempts(0);
+        DagManagerUtils.incrementJobGeneration(node);
+        Map<String, String> jobMetadata = TimingEventUtils.getJobMetadata(Maps.newHashMap(), node.getValue());
+        eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_PENDING_RESUME).stop(jobMetadata);
+      }
+      // Set flowStartTime so that flow SLA will be based on current time instead of original flow
+      node.getValue().setFlowStartTime(flowResumeTime);
+    }
+
+    dagManagementStateStore.checkpointDag(dag.get());
+    dagManagementStateStore.deleteFailedDag(dag.get());

Review Comment:
   does `checkpointDag` effectively add this to the dag state store` whereas `deleteFailedDag` removes it from the failed dag state store--basically a swap?
   
   a code comment would likely help maintainers



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that resumes a dag and submits the job that failed/killed previously.
+ */
+@Slf4j
+public class ResumeDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>> {
+
+  public ResumeDagProc(ResumeDagTask resumeDagTask) {
+    super(resumeDagTask);
+  }
+
+  @Override
+  protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore)
+      throws IOException {
+   return dagManagementStateStore.getFailedDag(getDagId());
+  }
+
+  @Override
+  protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag<JobExecutionPlan>> dag)
+      throws IOException {
+    log.info("Request to resume dag {}", getDagId());
+
+    if (!dag.isPresent()) {
+      // todo - add a metric here
+      log.error("Dag " + dagId + " was not found in dag state store");
+      return;
+    }
+
+    long flowResumeTime = System.currentTimeMillis();
+
+    // Set the flow and its failed or cancelled nodes to PENDING_RESUME so that the flow will be resumed from the point before it failed
+    DagManagerUtils.emitFlowEvent(eventSubmitter, dag.get(), TimingEvent.FlowTimings.FLOW_PENDING_RESUME);
+
+    for (Dag.DagNode<JobExecutionPlan> node : dag.get().getNodes()) {
+      ExecutionStatus executionStatus = node.getValue().getExecutionStatus();
+      if (executionStatus.equals(FAILED) || executionStatus.equals(CANCELLED)) {
+        node.getValue().setExecutionStatus(PENDING_RESUME);
+        // reset currentAttempts because we do not want to count previous execution's attempts in deciding whether to retry a job
+        node.getValue().setCurrentAttempts(0);
+        DagManagerUtils.incrementJobGeneration(node);
+        Map<String, String> jobMetadata = TimingEventUtils.getJobMetadata(Maps.newHashMap(), node.getValue());
+        eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_PENDING_RESUME).stop(jobMetadata);
+      }
+      // Set flowStartTime so that flow SLA will be based on current time instead of original flow

Review Comment:
   nit: flow *deadlines* (rather than "SLA")
   
   and to clarify, is this applicable to both job start deadline and flow completion deadline?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that resumes a dag and submits the job that failed/killed previously.

Review Comment:
   nit: "... submits the job that previously failed or was killed."



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that resumes a dag and submits the job that failed/killed previously.
+ */
+@Slf4j
+public class ResumeDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>> {
+
+  public ResumeDagProc(ResumeDagTask resumeDagTask) {
+    super(resumeDagTask);
+  }
+
+  @Override
+  protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore)
+      throws IOException {
+   return dagManagementStateStore.getFailedDag(getDagId());
+  }
+
+  @Override
+  protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag<JobExecutionPlan>> dag)
+      throws IOException {
+    log.info("Request to resume dag {}", getDagId());
+
+    if (!dag.isPresent()) {
+      // todo - add a metric here
+      log.error("Dag " + dagId + " was not found in dag state store");
+      return;
+    }
+
+    long flowResumeTime = System.currentTimeMillis();
+
+    // Set the flow and its failed or cancelled nodes to PENDING_RESUME so that the flow will be resumed from the point before it failed
+    DagManagerUtils.emitFlowEvent(eventSubmitter, dag.get(), TimingEvent.FlowTimings.FLOW_PENDING_RESUME);
+
+    for (Dag.DagNode<JobExecutionPlan> node : dag.get().getNodes()) {
+      ExecutionStatus executionStatus = node.getValue().getExecutionStatus();
+      if (executionStatus.equals(FAILED) || executionStatus.equals(CANCELLED)) {
+        node.getValue().setExecutionStatus(PENDING_RESUME);
+        // reset currentAttempts because we do not want to count previous execution's attempts in deciding whether to retry a job
+        node.getValue().setCurrentAttempts(0);
+        DagManagerUtils.incrementJobGeneration(node);

Review Comment:
   would be good to contextualize the currentAttempts "reset" with the "generation" increment



##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProcTest.java:
##########
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.api.FlowSpec;
+import org.apache.gobblin.runtime.api.Spec;
+import org.apache.gobblin.runtime.api.SpecProducer;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagActionStore;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.DagManagerTest;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.MostlyMySqlDagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.MostlyMySqlDagManagementStateStoreTest;
+import org.apache.gobblin.service.modules.orchestration.MysqlDagActionStore;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+
+
+public class ResumeDagProcTest {
+  private MostlyMySqlDagManagementStateStore dagManagementStateStore;
+  private ITestMetastoreDatabase testDb;
+
+  @BeforeClass
+  public void setUp() throws Exception {
+    testDb = TestMetastoreDatabaseFactory.get();
+    this.dagManagementStateStore = spy(MostlyMySqlDagManagementStateStoreTest.getDummyDMSS(testDb));
+    doReturn(FlowSpec.builder().build()).when(this.dagManagementStateStore).getFlowSpec(any());
+    doNothing().when(this.dagManagementStateStore).tryAcquireQuota(any());
+    doNothing().when(this.dagManagementStateStore).addDagNodeState(any(), any());
+  }
+
+  @AfterClass(alwaysRun = true)
+  public void tearDown() throws Exception {
+    if (testDb != null) {
+      testDb.close();
+    }
+  }
+
+  /*
+  This test creates a failed dag and launches a resume dag proc for it. It then verifies that the next jobs are set to run.
+   */
+  @Test
+  public void resumeDag() throws IOException, URISyntaxException {
+    long flowExecutionId = System.currentTimeMillis();
+    Dag<JobExecutionPlan> dag = DagManagerTest.buildDag("1", flowExecutionId, DagManager.FailureOption.FINISH_ALL_POSSIBLE.name(),
+        5, "user5", ConfigFactory.empty().withValue(ConfigurationKeys.FLOW_GROUP_KEY, ConfigValueFactory.fromAnyRef("fg")));
+    // simulate a failed dag in store
+    dag.getNodes().get(0).getValue().setExecutionStatus(ExecutionStatus.COMPLETE);
+    dag.getNodes().get(1).getValue().setExecutionStatus(ExecutionStatus.FAILED);
+    dag.getNodes().get(2).getValue().setExecutionStatus(ExecutionStatus.COMPLETE);
+    dag.getNodes().get(4).getValue().setExecutionStatus(ExecutionStatus.COMPLETE);
+    doReturn(Optional.of(dag)).when(dagManagementStateStore).getFailedDag(any());
+
+    ResumeDagProc resumeDagProc = new ResumeDagProc(new ResumeDagTask(new DagActionStore.DagAction("fg", "flow1",
+        String.valueOf(flowExecutionId), MysqlDagActionStore.NO_JOB_NAME_DEFAULT, DagActionStore.DagActionType.RESUME),
+        null, mock(DagActionStore.class)));
+    resumeDagProc.process(this.dagManagementStateStore);
+
+    List<SpecProducer<Spec>> specProducers = dag.getNodes().stream().map(n -> {
+      try {
+        return DagManagerUtils.getSpecProducer(n);
+      } catch (ExecutionException | InterruptedException e) {
+        throw new RuntimeException(e);
+      }
+    }).collect(Collectors.toList());
+    int expectedNumOfResumedJobs = 1; // = number of resumed nodes
+
+    long resumedJobCount = specProducers.stream()
+        .mapToLong(p -> Mockito.mockingDetails(p)
+            .getInvocations()
+            .stream()
+            .filter(a -> a.getMethod().getName().equals("addSpec"))
+            .count())
+        .sum();
+    long addedDagNodeStates = Mockito.mockingDetails(this.dagManagementStateStore).getInvocations().stream()
+        .filter(a -> a.getMethod().getName().equals("addDagNodeState")).count();
+
+    Assert.assertEquals(resumedJobCount, expectedNumOfResumedJobs);
+    Assert.assertEquals(addedDagNodeStates, expectedNumOfResumedJobs);

Review Comment:
   would the more common formulation not be equivalent?
   ```
   Mockito.verify(this.dagManagementStateStore, Mockito.times(expectedNumOfResumedJobs)).addDagNodeState(any(), any());
   ```



##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProcTest.java:
##########
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.api.FlowSpec;
+import org.apache.gobblin.runtime.api.Spec;
+import org.apache.gobblin.runtime.api.SpecProducer;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagActionStore;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.DagManagerTest;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.MostlyMySqlDagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.MostlyMySqlDagManagementStateStoreTest;
+import org.apache.gobblin.service.modules.orchestration.MysqlDagActionStore;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+
+
+public class ResumeDagProcTest {
+  private MostlyMySqlDagManagementStateStore dagManagementStateStore;
+  private ITestMetastoreDatabase testDb;
+
+  @BeforeClass
+  public void setUp() throws Exception {
+    testDb = TestMetastoreDatabaseFactory.get();
+    this.dagManagementStateStore = spy(MostlyMySqlDagManagementStateStoreTest.getDummyDMSS(testDb));
+    doReturn(FlowSpec.builder().build()).when(this.dagManagementStateStore).getFlowSpec(any());
+    doNothing().when(this.dagManagementStateStore).tryAcquireQuota(any());
+    doNothing().when(this.dagManagementStateStore).addDagNodeState(any(), any());
+  }
+
+  @AfterClass(alwaysRun = true)
+  public void tearDown() throws Exception {
+    if (testDb != null) {
+      testDb.close();
+    }
+  }
+
+  /*
+  This test creates a failed dag and launches a resume dag proc for it. It then verifies that the next jobs are set to run.
+   */
+  @Test
+  public void resumeDag() throws IOException, URISyntaxException {
+    long flowExecutionId = System.currentTimeMillis();
+    Dag<JobExecutionPlan> dag = DagManagerTest.buildDag("1", flowExecutionId, DagManager.FailureOption.FINISH_ALL_POSSIBLE.name(),
+        5, "user5", ConfigFactory.empty().withValue(ConfigurationKeys.FLOW_GROUP_KEY, ConfigValueFactory.fromAnyRef("fg")));
+    // simulate a failed dag in store
+    dag.getNodes().get(0).getValue().setExecutionStatus(ExecutionStatus.COMPLETE);
+    dag.getNodes().get(1).getValue().setExecutionStatus(ExecutionStatus.FAILED);
+    dag.getNodes().get(2).getValue().setExecutionStatus(ExecutionStatus.COMPLETE);
+    dag.getNodes().get(4).getValue().setExecutionStatus(ExecutionStatus.COMPLETE);
+    doReturn(Optional.of(dag)).when(dagManagementStateStore).getFailedDag(any());
+
+    ResumeDagProc resumeDagProc = new ResumeDagProc(new ResumeDagTask(new DagActionStore.DagAction("fg", "flow1",

Review Comment:
   let's give magic strings a `static final` name
   
   ...and was "flow1" shown above as "1" (on line 87)?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["phet (via GitHub)" <gi...@apache.org>]

phet commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1594903505


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that resumes a dag and submits the job that failed/killed previously.
+ */
+@Slf4j
+public class ResumeDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>> {
+
+  public ResumeDagProc(ResumeDagTask resumeDagTask) {
+    super(resumeDagTask);
+  }
+
+  @Override
+  protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore)
+      throws IOException {
+   return dagManagementStateStore.getFailedDag(getDagId());
+  }
+
+  @Override
+  protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag<JobExecutionPlan>> dag)
+      throws IOException {
+    log.info("Request to resume dag {}", getDagId());
+
+    if (!dag.isPresent()) {
+      // todo - add a metric here
+      log.error("Dag " + dagId + " was not found in dag state store");
+      return;
+    }
+
+    long flowResumeTime = System.currentTimeMillis();
+
+    // Set the flow and its failed or cancelled nodes to PENDING_RESUME so that the flow will be resumed from the point before it failed
+    DagManagerUtils.emitFlowEvent(eventSubmitter, dag.get(), TimingEvent.FlowTimings.FLOW_PENDING_RESUME);
+
+    for (Dag.DagNode<JobExecutionPlan> node : dag.get().getNodes()) {
+      ExecutionStatus executionStatus = node.getValue().getExecutionStatus();
+      if (executionStatus.equals(FAILED) || executionStatus.equals(CANCELLED)) {
+        node.getValue().setExecutionStatus(PENDING_RESUME);
+        // reset currentAttempts because we do not want to count previous execution's attempts in deciding whether to retry a job
+        node.getValue().setCurrentAttempts(0);
+        DagManagerUtils.incrementJobGeneration(node);
+        Map<String, String> jobMetadata = TimingEventUtils.getJobMetadata(Maps.newHashMap(), node.getValue());
+        eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_PENDING_RESUME).stop(jobMetadata);
+      }
+      // Set flowStartTime so that flow SLA will be based on current time instead of original flow
+      node.getValue().setFlowStartTime(flowResumeTime);
+    }
+
+    dagManagementStateStore.checkpointDag(dag.get());
+    dagManagementStateStore.deleteFailedDag(dag.get());

Review Comment:
   does `checkpointDag` effectively add this to the dag state store whereas `deleteFailedDag` removes it from the failed dag state store--basically a swap?
   
   a code comment would likely help maintainers



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["phet (via GitHub)" <gi...@apache.org>]

phet commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1596012740


##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProcTest.java:
##########
@@ -84,16 +84,19 @@ public void tearDown() throws Exception {
   @Test
   public void resumeDag() throws IOException, URISyntaxException {
     long flowExecutionId = System.currentTimeMillis();

Review Comment:
   is there any reason NOT to just use a constant here?



##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -82,12 +82,14 @@ protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag
         Map<String, String> jobMetadata = TimingEventUtils.getJobMetadata(Maps.newHashMap(), node.getValue());
         eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_PENDING_RESUME).stop(jobMetadata);
       }
-      // Set flowStartTime so that flow SLA will be based on current time instead of original flow
+      // Set flowStartTime so that flow start deadline and flow completion deadline will be based on current time instead of original flow
       node.getValue().setFlowStartTime(flowResumeTime);
     }
 
+    // these two statements effectively move the dag from failed dag store to (running) dag store

Review Comment:
   maybe add - "to prevent loss in the unlikely event of failure between the two, we add first"
   
   that said, should we check for this state around L68/69?



##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProcTest.java:
##########
@@ -114,10 +117,8 @@ public void resumeDag() throws IOException, URISyntaxException {
             .filter(a -> a.getMethod().getName().equals("addSpec"))
             .count())
         .sum();
-    long addedDagNodeStates = Mockito.mockingDetails(this.dagManagementStateStore).getInvocations().stream()
-        .filter(a -> a.getMethod().getName().equals("addDagNodeState")).count();
 
     Assert.assertEquals(resumedJobCount, expectedNumOfResumedJobs);

Review Comment:
   can't we similarly use `Mockito.verify` and `Mockito.times` here rather than `getInvocations()` and the `.filter` + `.count`?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["phet (via GitHub)" <gi...@apache.org>]

phet commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1594904360


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that resumes a dag and submits the job that failed/killed previously.
+ */
+@Slf4j
+public class ResumeDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>> {
+
+  public ResumeDagProc(ResumeDagTask resumeDagTask) {
+    super(resumeDagTask);
+  }
+
+  @Override
+  protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore)
+      throws IOException {
+   return dagManagementStateStore.getFailedDag(getDagId());
+  }
+
+  @Override
+  protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag<JobExecutionPlan>> dag)
+      throws IOException {
+    log.info("Request to resume dag {}", getDagId());
+
+    if (!dag.isPresent()) {
+      // todo - add a metric here
+      log.error("Dag " + dagId + " was not found in dag state store");
+      return;
+    }
+
+    long flowResumeTime = System.currentTimeMillis();
+
+    // Set the flow and its failed or cancelled nodes to PENDING_RESUME so that the flow will be resumed from the point before it failed
+    DagManagerUtils.emitFlowEvent(eventSubmitter, dag.get(), TimingEvent.FlowTimings.FLOW_PENDING_RESUME);
+
+    for (Dag.DagNode<JobExecutionPlan> node : dag.get().getNodes()) {
+      ExecutionStatus executionStatus = node.getValue().getExecutionStatus();
+      if (executionStatus.equals(FAILED) || executionStatus.equals(CANCELLED)) {
+        node.getValue().setExecutionStatus(PENDING_RESUME);
+        // reset currentAttempts because we do not want to count previous execution's attempts in deciding whether to retry a job
+        node.getValue().setCurrentAttempts(0);
+        DagManagerUtils.incrementJobGeneration(node);
+        Map<String, String> jobMetadata = TimingEventUtils.getJobMetadata(Maps.newHashMap(), node.getValue());
+        eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_PENDING_RESUME).stop(jobMetadata);
+      }
+      // Set flowStartTime so that flow SLA will be based on current time instead of original flow
+      node.getValue().setFlowStartTime(flowResumeTime);
+    }
+
+    dagManagementStateStore.checkpointDag(dag.get());
+    dagManagementStateStore.deleteFailedDag(dag.get());
+    resumeDag(dagManagementStateStore, dag.get());
+  }
+
+  private void resumeDag(DagManagementStateStore dagManagementStateStore, Dag<JobExecutionPlan> dag) {
+    Set<Dag.DagNode<JobExecutionPlan>> nextNodes = DagManagerUtils.getNext(dag);
+
+    if (nextNodes.size() > 1) {
+      handleMultipleJobs(nextNodes);
+    }
+
+    //Submit jobs from the dag ready for execution.
+    for (Dag.DagNode<JobExecutionPlan> dagNode : nextNodes) {
+      DagProcUtils.submitJobToExecutor(dagManagementStateStore, dagNode, getDagId());
+      log.info("Submitted job {} for dagId {}", DagManagerUtils.getJobName(dagNode), getDagId());
+    }
+  }
+
+  private void handleMultipleJobs(Set<Dag.DagNode<JobExecutionPlan>> nextNodes) {
+    throw new UnsupportedOperationException("More than one start job is not allowed");
+  }

Review Comment:
   once this PR is merged, I believe we'll be ready to implement this (all three cases)!



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["arjun4084346 (via GitHub)" <gi...@apache.org>]

arjun4084346 commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1596027289


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -82,12 +82,14 @@ protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag
         Map<String, String> jobMetadata = TimingEventUtils.getJobMetadata(Maps.newHashMap(), node.getValue());
         eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_PENDING_RESUME).stop(jobMetadata);
       }
-      // Set flowStartTime so that flow SLA will be based on current time instead of original flow
+      // Set flowStartTime so that flow start deadline and flow completion deadline will be based on current time instead of original flow
       node.getValue().setFlowStartTime(flowResumeTime);
     }
 
+    // these two statements effectively move the dag from failed dag store to (running) dag store

Review Comment:
   check for what state?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["arjun4084346 (via GitHub)" <gi...@apache.org>]

arjun4084346 commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1596027618


##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProcTest.java:
##########
@@ -84,16 +84,19 @@ public void tearDown() throws Exception {
   @Test
   public void resumeDag() throws IOException, URISyntaxException {
     long flowExecutionId = System.currentTimeMillis();

Review Comment:
   no reason



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["arjun4084346 (via GitHub)" <gi...@apache.org>]

arjun4084346 commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1595694050


##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagManagementDagActionStoreChangeMonitor.java:
##########
@@ -63,10 +63,11 @@ protected void handleDagAction(DagActionStore.DagAction dagAction, boolean isSta
         case LAUNCH :
         case REEVALUATE :
         case KILL :
+        case RESUME:

Review Comment:
   i was listing them in the order of execution, which is also same as order of occurrence.
   first a dag is launched, then reevaluated, then killed or resumed



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["phet (via GitHub)" <gi...@apache.org>]

phet commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1596104629


##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProcTest.java:
##########
@@ -114,10 +117,8 @@ public void resumeDag() throws IOException, URISyntaxException {
             .filter(a -> a.getMethod().getName().equals("addSpec"))
             .count())
         .sum();
-    long addedDagNodeStates = Mockito.mockingDetails(this.dagManagementStateStore).getInvocations().stream()
-        .filter(a -> a.getMethod().getName().equals("addDagNodeState")).count();
 
     Assert.assertEquals(resumedJobCount, expectedNumOfResumedJobs);

Review Comment:
   wouldn't that actually be better to verify that the spec producer associated w/ the `DagNode` we expect is the specific one with its `.addSpec` invoked?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["arjun4084346 (via GitHub)" <gi...@apache.org>]

arjun4084346 merged PR #3938:
URL: https://github.com/apache/gobblin/pull/3938


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["umustafi (via GitHub)" <gi...@apache.org>]

umustafi commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1595836249


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that resumes a dag and submits the job that previously failed or was killed.
+ */
+@Slf4j
+public class ResumeDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>> {
+
+  public ResumeDagProc(ResumeDagTask resumeDagTask) {
+    super(resumeDagTask);
+  }
+
+  @Override
+  protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore)
+      throws IOException {
+   return dagManagementStateStore.getFailedDag(getDagId());
+  }
+
+  @Override
+  protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag<JobExecutionPlan>> dag)
+      throws IOException {
+    log.info("Request to resume dag {}", getDagId());
+
+    if (!dag.isPresent()) {
+      // todo - add a metric here
+      log.error("Dag " + dagId + " was not found in dag state store");
+      return;
+    }
+
+    long flowResumeTime = System.currentTimeMillis();
+
+    // Set the flow and its failed or cancelled nodes to PENDING_RESUME so that the flow will be resumed from the point before it failed
+    DagManagerUtils.emitFlowEvent(eventSubmitter, dag.get(), TimingEvent.FlowTimings.FLOW_PENDING_RESUME);
+
+    for (Dag.DagNode<JobExecutionPlan> node : dag.get().getNodes()) {
+      ExecutionStatus executionStatus = node.getValue().getExecutionStatus();
+      if (executionStatus.equals(FAILED) || executionStatus.equals(CANCELLED)) {
+        node.getValue().setExecutionStatus(PENDING_RESUME);
+        // reset currentAttempts because we do not want to count previous execution's attempts in deciding whether to retry a job
+        node.getValue().setCurrentAttempts(0);
+        DagManagerUtils.incrementJobGeneration(node);
+        Map<String, String> jobMetadata = TimingEventUtils.getJobMetadata(Maps.newHashMap(), node.getValue());
+        eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_PENDING_RESUME).stop(jobMetadata);
+      }
+      // Set flowStartTime so that flow start deadline and flow completion deadline will be based on current time instead of original flow
+      node.getValue().setFlowStartTime(flowResumeTime);
+    }
+
+    // these two statements effectively move the dag from failed dag store to (running) dag store
+    dagManagementStateStore.checkpointDag(dag.get());
+    dagManagementStateStore.deleteFailedDag(dag.get());
+
+    resumeDag(dagManagementStateStore, dag.get());
+  }
+
+  private void resumeDag(DagManagementStateStore dagManagementStateStore, Dag<JobExecutionPlan> dag) {
+    Set<Dag.DagNode<JobExecutionPlan>> nextNodes = DagManagerUtils.getNext(dag);
+
+    if (nextNodes.size() > 1) {
+      handleMultipleJobs(nextNodes);
+    }
+
+    //Submit jobs from the dag ready for execution.
+    for (Dag.DagNode<JobExecutionPlan> dagNode : nextNodes) {
+      DagProcUtils.submitJobToExecutor(dagManagementStateStore, dagNode, getDagId());
+      log.info("Submitted job {} for dagId {}", DagManagerUtils.getJobName(dagNode), getDagId());
+    }
+  }

Review Comment:
   in future if there are multiple next jobs are we putting them each as a separate launch_job action in the store? then we will not need to loop through `nextNodes` right?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["arjun4084346 (via GitHub)" <gi...@apache.org>]

arjun4084346 commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1596032031


##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProc.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration.proc;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.collect.Maps;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.orchestration.DagManagementStateStore;
+import org.apache.gobblin.service.modules.orchestration.DagManagerUtils;
+import org.apache.gobblin.service.modules.orchestration.TimingEventUtils;
+import org.apache.gobblin.service.modules.orchestration.task.ResumeDagTask;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+
+import static org.apache.gobblin.service.ExecutionStatus.CANCELLED;
+import static org.apache.gobblin.service.ExecutionStatus.FAILED;
+import static org.apache.gobblin.service.ExecutionStatus.PENDING_RESUME;
+
+
+/**
+ * An implementation for {@link DagProc} that resumes a dag and submits the job that previously failed or was killed.
+ */
+@Slf4j
+public class ResumeDagProc extends DagProc<Optional<Dag<JobExecutionPlan>>> {
+
+  public ResumeDagProc(ResumeDagTask resumeDagTask) {
+    super(resumeDagTask);
+  }
+
+  @Override
+  protected Optional<Dag<JobExecutionPlan>> initialize(DagManagementStateStore dagManagementStateStore)
+      throws IOException {
+   return dagManagementStateStore.getFailedDag(getDagId());
+  }
+
+  @Override
+  protected void act(DagManagementStateStore dagManagementStateStore, Optional<Dag<JobExecutionPlan>> dag)
+      throws IOException {
+    log.info("Request to resume dag {}", getDagId());
+
+    if (!dag.isPresent()) {
+      // todo - add a metric here
+      log.error("Dag " + dagId + " was not found in dag state store");
+      return;
+    }
+
+    long flowResumeTime = System.currentTimeMillis();
+
+    // Set the flow and its failed or cancelled nodes to PENDING_RESUME so that the flow will be resumed from the point before it failed
+    DagManagerUtils.emitFlowEvent(eventSubmitter, dag.get(), TimingEvent.FlowTimings.FLOW_PENDING_RESUME);
+
+    for (Dag.DagNode<JobExecutionPlan> node : dag.get().getNodes()) {
+      ExecutionStatus executionStatus = node.getValue().getExecutionStatus();
+      if (executionStatus.equals(FAILED) || executionStatus.equals(CANCELLED)) {
+        node.getValue().setExecutionStatus(PENDING_RESUME);
+        // reset currentAttempts because we do not want to count previous execution's attempts in deciding whether to retry a job
+        node.getValue().setCurrentAttempts(0);
+        DagManagerUtils.incrementJobGeneration(node);
+        Map<String, String> jobMetadata = TimingEventUtils.getJobMetadata(Maps.newHashMap(), node.getValue());
+        eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_PENDING_RESUME).stop(jobMetadata);
+      }
+      // Set flowStartTime so that flow start deadline and flow completion deadline will be based on current time instead of original flow
+      node.getValue().setFlowStartTime(flowResumeTime);
+    }
+
+    // these two statements effectively move the dag from failed dag store to (running) dag store
+    dagManagementStateStore.checkpointDag(dag.get());
+    dagManagementStateStore.deleteFailedDag(dag.get());
+
+    resumeDag(dagManagementStateStore, dag.get());
+  }
+
+  private void resumeDag(DagManagementStateStore dagManagementStateStore, Dag<JobExecutionPlan> dag) {
+    Set<Dag.DagNode<JobExecutionPlan>> nextNodes = DagManagerUtils.getNext(dag);
+
+    if (nextNodes.size() > 1) {
+      handleMultipleJobs(nextNodes);
+    }
+
+    //Submit jobs from the dag ready for execution.
+    for (Dag.DagNode<JobExecutionPlan> dagNode : nextNodes) {
+      DagProcUtils.submitJobToExecutor(dagManagementStateStore, dagNode, getDagId());
+      log.info("Submitted job {} for dagId {}", DagManagerUtils.getJobName(dagNode), getDagId());
+    }
+  }

Review Comment:
   right



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["arjun4084346 (via GitHub)" <gi...@apache.org>]

arjun4084346 commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1596030629


##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProcTest.java:
##########
@@ -114,10 +117,8 @@ public void resumeDag() throws IOException, URISyntaxException {
             .filter(a -> a.getMethod().getName().equals("addSpec"))
             .count())
         .sum();
-    long addedDagNodeStates = Mockito.mockingDetails(this.dagManagementStateStore).getInvocations().stream()
-        .filter(a -> a.getMethod().getName().equals("addDagNodeState")).count();
 
     Assert.assertEquals(resumedJobCount, expectedNumOfResumedJobs);

Review Comment:
   there are multiple objects on which invocations is to be find and summed. not sure if a direct mockito method provides that functionality



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [GOBBLIN-2057]add resume dag proc [gobblin]

["phet (via GitHub)" <gi...@apache.org>]

phet commented on code in PR #3938:
URL: https://github.com/apache/gobblin/pull/3938#discussion_r1596104629


##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/ResumeDagProcTest.java:
##########
@@ -114,10 +117,8 @@ public void resumeDag() throws IOException, URISyntaxException {
             .filter(a -> a.getMethod().getName().equals("addSpec"))
             .count())
         .sum();
-    long addedDagNodeStates = Mockito.mockingDetails(this.dagManagementStateStore).getInvocations().stream()
-        .filter(a -> a.getMethod().getName().equals("addDagNodeState")).count();
 
     Assert.assertEquals(resumedJobCount, expectedNumOfResumedJobs);

Review Comment:
   wouldn't that even be best to check that the spec producer associated w/ the `DagNode` we expect it for is the specific one with its `.addSpec` invoked?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org