You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by di...@apache.org on 2019/03/01 20:50:02 UTC
[airavata] branch staging updated: Setting force run mode for
Airavata Tasks
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch staging
in repository https://gitbox.apache.org/repos/asf/airavata.git
The following commit(s) were added to refs/heads/staging by this push:
new 0f0a52a Setting force run mode for Airavata Tasks
0f0a52a is described below
commit 0f0a52afadcb9bc33439cfb6be4ceb062a01ebfa
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Fri Mar 1 15:49:52 2019 -0500
Setting force run mode for Airavata Tasks
---
.../airavata/helix/impl/task/AiravataTask.java | 21 +++++++++++++++++++++
.../airavata/helix/impl/task/TaskContext.java | 5 +++--
.../helix/impl/workflow/PostWorkflowManager.java | 6 ++++--
.../helix/impl/workflow/PreWorkflowManager.java | 5 +++--
4 files changed, 31 insertions(+), 6 deletions(-)
diff --git a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/AiravataTask.java b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/AiravataTask.java
index 8c7cd7e..4c8adad 100644
--- a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/AiravataTask.java
+++ b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/AiravataTask.java
@@ -81,6 +81,9 @@ public abstract class AiravataTask extends AbstractTask {
@TaskParam(name = "Skip Status Publish")
private boolean skipTaskStatusPublish = false;
+ @TaskParam(name ="Force Run Task")
+ private boolean forceRunTask = false;
+
protected TaskResult onSuccess(String message) {
logger.info(message);
if (!skipTaskStatusPublish) {
@@ -339,6 +342,16 @@ public abstract class AiravataTask extends AbstractTask {
MDC.put("gateway", getGatewayId());
MDC.put("task", getTaskId());
loadContext();
+ if (!forceRunTask) {
+ if (this.taskContext != null) {
+ TaskState taskState = taskContext.getTaskState();
+ if (taskState != null && taskState != TaskState.CREATED) {
+ logger.warn("Task " + getTaskId() + " is not in CREATED state. So skipping execution");
+ skipTaskStatusPublish = false;
+ return onSuccess("Task " + getTaskId() + " is not in CREATED state. So skipping execution");
+ }
+ }
+ }
if (!skipTaskStatusPublish) {
publishTaskState(TaskState.EXECUTING);
}
@@ -489,6 +502,14 @@ public abstract class AiravataTask extends AbstractTask {
return skipTaskStatusPublish;
}
+ public boolean isForceRunTask() {
+ return forceRunTask;
+ }
+
+ public void setForceRunTask(boolean forceRunTask) {
+ this.forceRunTask = forceRunTask;
+ }
+
// TODO this is inefficient. Try to use a connection pool
public static RegistryService.Client getRegistryServiceClient() {
try {
diff --git a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/TaskContext.java b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/TaskContext.java
index f2ec7c8..3192a59 100644
--- a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/TaskContext.java
+++ b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/task/TaskContext.java
@@ -394,10 +394,11 @@ public class TaskContext {
}
public TaskState getTaskState() {
- if(getCurrentTaskModel().getTaskStatuses() != null)
+ if(getCurrentTaskModel() != null && getCurrentTaskModel().getTaskStatuses() != null) {
return getCurrentTaskModel().getTaskStatuses().get(0).getState();
- else
+ } else {
return null;
+ }
}
public TaskStatus getTaskStatus() {
diff --git a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PostWorkflowManager.java b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PostWorkflowManager.java
index afff17d..cb71ef5 100644
--- a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PostWorkflowManager.java
+++ b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PostWorkflowManager.java
@@ -160,7 +160,7 @@ public class PostWorkflowManager extends WorkflowManager {
logger.info("Job " + jobStatusResult.getJobId() + " was completed");
- executePostWorkflow(processId, gateway);
+ executePostWorkflow(processId, gateway, false);
} else if (jobStatusResult.getState() == JobState.CANCELED) {
logger.info("Job " + jobStatusResult.getJobId() + " was externally cancelled but process is not marked as cancelled yet");
@@ -184,7 +184,7 @@ public class PostWorkflowManager extends WorkflowManager {
}
}
- private void executePostWorkflow(String processId, String gateway) throws Exception {
+ private void executePostWorkflow(String processId, String gateway, boolean forceRun) throws Exception {
RegistryService.Client registryClient = getRegistryClientPool().getResource();
@@ -238,6 +238,7 @@ public class PostWorkflowManager extends WorkflowManager {
airavataTask.setProcessId(processModel.getProcessId());
airavataTask.setTaskId(taskModel.getTaskId());
airavataTask.setRetryCount(taskModel.getMaxRetry());
+ airavataTask.setForceRunTask(forceRun);
if (allTasks.size() > 0) {
allTasks.get(allTasks.size() - 1).setNextTask(new OutPort(airavataTask.getTaskId(), airavataTask));
}
@@ -251,6 +252,7 @@ public class PostWorkflowManager extends WorkflowManager {
completingTask.setExperimentId(experimentModel.getExperimentId());
completingTask.setProcessId(processModel.getProcessId());
completingTask.setTaskId("Completing-Task");
+ completingTask.setForceRunTask(forceRun);
completingTask.setSkipTaskStatusPublish(true);
if (allTasks.size() > 0) {
allTasks.get(allTasks.size() - 1).setNextTask(new OutPort(completingTask.getTaskId(), completingTask));
diff --git a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PreWorkflowManager.java b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PreWorkflowManager.java
index 3bac00b..15d6735 100644
--- a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PreWorkflowManager.java
+++ b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PreWorkflowManager.java
@@ -71,7 +71,7 @@ public class PreWorkflowManager extends WorkflowManager {
this.subscriber = MessagingFactory.getSubscriber(new ProcessLaunchMessageHandler(), routingKeys, Type.PROCESS_LAUNCH);
}
- private String createAndLaunchPreWorkflow(String processId) throws Exception {
+ private String createAndLaunchPreWorkflow(String processId, boolean forceRun) throws Exception {
RegistryService.Client registryClient = getRegistryClientPool().getResource();
@@ -119,6 +119,7 @@ public class PreWorkflowManager extends WorkflowManager {
airavataTask.setProcessId(processModel.getProcessId());
airavataTask.setTaskId(taskModel.getTaskId());
airavataTask.setRetryCount(taskModel.getMaxRetry());
+ airavataTask.setForceRunTask(forceRun);
if (allTasks.size() > 0) {
allTasks.get(allTasks.size() -1).setNextTask(new OutPort(airavataTask.getTaskId(), airavataTask));
}
@@ -232,7 +233,7 @@ public class PreWorkflowManager extends WorkflowManager {
try {
logger.info("Launching the pre workflow for process " + processId + " of experiment " + experimentId + " in gateway " + gateway);
- String workflowName = createAndLaunchPreWorkflow(processId);
+ String workflowName = createAndLaunchPreWorkflow(processId, false);
logger.info("Completed launching the pre workflow " + workflowName + " for process" + processId + " of experiment " + experimentId + " in gateway " + gateway);
// updating the process status