Refactoring post workflow manager
@@ -140,12 +140,37 @@ public class PostWorkflowManager extends WorkflowManager {
if (jobStatusResult.getState() == JobState.COMPLETE || jobStatusResult.getState() == JobState.FAILED) {
// if the job is FAILED, still run output staging tasks to debug the reason for failure. And update
// the experiment status as COMPLETED as this job failure is not related to Airavata scope.
-
logger.info("Starting the post workflow for job id : " + jobStatusResult.getJobId() + " with process id "
+ processId + ", gateway " + gateway + " and status " + jobStatusResult.getState().name());
logger.info("Job " + jobStatusResult.getJobId() + " was completed");
+ executePostWorkflow(processId, gateway);
+
+ } else if (jobStatusResult.getState() == JobState.CANCELED) {
+ logger.info("Job " + jobStatusResult.getJobId() + " was externally cancelled but process is not marked as cancelled yet");
+ MonitoringUtil.registerCancelProcess(getCuratorClient(), processId);
+ publishProcessStatus(processId, experimentId,gateway, ProcessState.CANCELED);
+ logger.info("Marked process " + processId + " of experiment " + experimentId + " as cancelled as job is already being cancelled");
+
+ } else if (jobStatusResult.getState() == JobState.SUBMITTED) {
+ logger.info("Job " + jobStatusResult.getJobId() + " was submitted");
+
+ }
+ }
+ return true;
+ } else {
+ logger.warn("Could not find a monitoring register for job id " + jobStatusResult.getJobId());
+ return false;
+ }
+ } catch (Exception e) {
+ logger.error("Failed to process job : " + jobStatusResult.getJobId() + ", with status : " + jobStatusResult.getState().name(), e);
+ return false;
+ }
+ }
+
+ private void executePostWorkflow(String processId, String gateway) throws Exception {
+
RegistryService.Client registryClient = getRegistryClientPool().getResource();
ProcessModel processModel;
@@ -225,26 +250,6 @@ public class PostWorkflowManager extends WorkflowManager {
"This will affect cancellation tasks", e);
}
- } else if (jobStatusResult.getState() == JobState.CANCELED) {
- logger.info("Job " + jobStatusResult.getJobId() + " was externally cancelled but process is not marked as cancelled yet");
- MonitoringUtil.registerCancelProcess(getCuratorClient(), processId);
- publishProcessStatus(processId, experimentId,gateway, ProcessState.CANCELED);
- logger.info("Marked process " + processId + " of experiment " + experimentId + " as cancelled as job is already being cancelled");
-
- } else if (jobStatusResult.getState() == JobState.SUBMITTED) {
- logger.info("Job " + jobStatusResult.getJobId() + " was submitted");
-
- }
- }
- return true;
- } else {
- logger.warn("Could not find a monitoring register for job id " + jobStatusResult.getJobId());
- return false;
- }
- } catch (Exception e) {
- logger.error("Failed to process job : " + jobStatusResult.getJobId() + ", with status : " + jobStatusResult.getState().name(), e);
- return false;
- }
}
private void runConsumer() throws ApplicationSettingsException {
