Adding new task repositories
ADD
modules/registry-refactoring/src/main/java/org/apache/airavata/registry/core/repositories/expcatalog/TaskErrorRepository.java
+package org.apache.airavata.registry.core.repositories.expcatalog;
+
+import org.apache.airavata.model.commons.ErrorModel;
+import org.apache.airavata.model.task.TaskModel;
+import org.apache.airavata.registry.core.entities.expcatalog.TaskErrorEntity;
+import org.apache.airavata.registry.core.entities.expcatalog.TaskErrorPK;
+import org.apache.airavata.registry.core.utils.ExpCatalogUtils;
+import org.apache.airavata.registry.core.utils.ObjectMapperSingleton;
+import org.apache.airavata.registry.cpi.RegistryException;
+import org.dozer.Mapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+public class TaskErrorRepository extends ExpCatAbstractRepository<ErrorModel, TaskErrorEntity, TaskErrorPK> {
+ private final static Logger logger = LoggerFactory.getLogger(TaskErrorRepository.class);
+
+ public TaskErrorRepository() { super(ErrorModel.class, TaskErrorEntity.class); }
+
+ protected String saveTaskError(ErrorModel error, String taskId) throws RegistryException {
+ Mapper mapper = ObjectMapperSingleton.getInstance();
+ TaskErrorEntity taskErrorEntity = mapper.map(error, TaskErrorEntity.class);
+
+ if (taskErrorEntity.getTaskId() == null) {
+ logger.debug("Setting the TaskErrorEntity's TaskId");
+ taskErrorEntity.setTaskId(taskId);
+ }
+
+ execute(entityManager -> entityManager.merge(taskErrorEntity));
+ return taskErrorEntity.getErrorId();
+ }
+
+ public String addTaskError(ErrorModel taskError, String taskId) throws RegistryException {
+
+ if (taskError.getErrorId() == null) {
+ logger.debug("Setting the TaskError's ErrorId");
+ taskError.setErrorId(ExpCatalogUtils.getID("ERROR"));
+ }
+
+ return saveTaskError(taskError, taskId);
+ }
+
+ public String updateTaskError(ErrorModel updatedTaskError, String taskId) throws RegistryException {
+ return saveTaskError(updatedTaskError, taskId);
+ }
+
+ public List<ErrorModel> getTaskError(String taskId) throws RegistryException {
+ TaskRepository taskRepository = new TaskRepository();
+ TaskModel taskModel = taskRepository.getTask(taskId);
+ return taskModel.getTaskErrors();
+ }
+
+}

MODIFY
modules/registry-refactoring/src/main/java/org/apache/airavata/registry/core/repositories/expcatalog/TaskRepository.java
modules/registry-refactoring/src/main/java/org/apache/airavata/registry/core/repositories/expcatalog/TaskRepository.java
*/
package org.apache.airavata.registry.core.repositories.expcatalog;
-import org.apache.airavata.model.commons.ErrorModel;
import org.apache.airavata.model.commons.airavata_commonsConstants;
-import org.apache.airavata.model.status.TaskState;
-import org.apache.airavata.model.status.TaskStatus;
import org.apache.airavata.model.task.TaskModel;
import org.apache.airavata.registry.core.entities.expcatalog.TaskEntity;
import org.apache.airavata.registry.core.utils.DBConstants;
@@ -97,101 +94,6 @@ public class TaskRepository extends ExpCatAbstractRepository<TaskModel, TaskEnti
return taskRepository.get(taskId);
}
- public String addTaskStatus(TaskStatus taskStatus, String taskId) throws RegistryException {
- TaskModel taskModel = getTask(taskId);
- List<TaskStatus> taskStatusList = taskModel.getTaskStatuses();
-
- if (taskStatusList.size() == 0 || !taskStatusList.contains(taskStatus)) {
-
- if (taskStatus.getStatusId() == null) {
- logger.debug("Set TaskStatus's StatusId");
- taskStatus.setStatusId(ExpCatalogUtils.getID("STATUS"));
- }
-
- logger.debug("Adding the TaskStatus to the list");
- taskStatusList.add(taskStatus);
- }
-
- taskModel.setTaskStatuses(taskStatusList);
- updateTask(taskModel, taskId);
- return taskStatus.getStatusId();
- }
-
- public String updateTaskStatus(TaskStatus updatedTaskStatus, String taskId) throws RegistryException {
- TaskModel taskModel = getTask(taskId);
- List<TaskStatus> taskStatusList = taskModel.getTaskStatuses();
-
- for (TaskStatus retrievedTaskStatus : taskStatusList) {
-
- if (retrievedTaskStatus.getStatusId().equals(updatedTaskStatus.getStatusId())) {
- logger.debug("Updating the TaskStatus");
- taskStatusList.remove(retrievedTaskStatus);
- taskStatusList.add(updatedTaskStatus);
- }
-
- }
-
- taskModel.setTaskStatuses(taskStatusList);
- updateTask(taskModel, taskId);
- return updatedTaskStatus.getStatusId();
- }
-
- public TaskStatus getTaskStatus(String taskId) throws RegistryException {
- TaskModel taskModel = getTask(taskId);
- List<TaskStatus> taskStatusList = taskModel.getTaskStatuses();
-
- if(taskStatusList.size() == 0) {
- logger.debug("TaskStatus list is empty");
- return null;
- }
-
- else {
- TaskStatus latestTaskStatus = taskStatusList.get(0);
-
- for(int i = 1; i < taskStatusList.size(); i++) {
- Timestamp timeOfStateChange = new Timestamp(taskStatusList.get(i).getTimeOfStateChange());
-
- if (timeOfStateChange.after(new Timestamp(latestTaskStatus.getTimeOfStateChange()))
- || (timeOfStateChange.equals(latestTaskStatus.getTimeOfStateChange()) && taskStatusList.get(i).getState().equals(TaskState.COMPLETED.toString()))
- || (timeOfStateChange.equals(latestTaskStatus.getTimeOfStateChange()) && taskStatusList.get(i).getState().equals(TaskState.FAILED.toString()))
- || (timeOfStateChange.equals(latestTaskStatus.getTimeOfStateChange()) && taskStatusList.get(i).getState().equals(TaskState.CANCELED.toString()))) {
- latestTaskStatus = taskStatusList.get(i);
- }
-
- }
-
- return latestTaskStatus;
- }
- }
-
- public String addTaskError(ErrorModel taskError, String taskId) throws RegistryException {
- TaskModel taskModel = getTask(taskId);
- List<ErrorModel> errorModelList = taskModel.getTaskErrors();
-
- if (errorModelList == null) {
- logger.debug("Adding the first TaskError to the list");
- taskModel.setTaskErrors(Arrays.asList(taskError));
- }
-
- else if (!errorModelList.contains(taskError)) {
- logger.debug("Adding the TaskError to the list");
- errorModelList.add(taskError);
- taskModel.setTaskErrors(errorModelList);
- }
-
- updateTask(taskModel, taskId);
- return taskId;
- }
-
- public String updateTaskError(ErrorModel taskError, String taskId) throws RegistryException {
- return addTaskError(taskError, taskId);
- }
-
- public List<ErrorModel> getTaskError(String taskId) throws RegistryException {
- TaskModel taskModel = getTask(taskId);
- return taskModel.getTaskErrors();
- }
-
public List<TaskModel> getTaskList(String fieldName, Object value) throws RegistryException {
TaskRepository taskRepository = new TaskRepository();
List<TaskModel> taskModelList;

ADD
modules/registry-refactoring/src/main/java/org/apache/airavata/registry/core/repositories/expcatalog/TaskStatusRepository.java
+package org.apache.airavata.registry.core.repositories.expcatalog;
+
+import org.apache.airavata.model.status.TaskState;
+import org.apache.airavata.model.status.TaskStatus;
+import org.apache.airavata.model.task.TaskModel;
+import org.apache.airavata.registry.core.entities.expcatalog.TaskStatusEntity;
+import org.apache.airavata.registry.core.entities.expcatalog.TaskStatusPK;
+import org.apache.airavata.registry.core.utils.ExpCatalogUtils;
+import org.apache.airavata.registry.core.utils.ObjectMapperSingleton;
+import org.apache.airavata.registry.cpi.RegistryException;
+import org.dozer.Mapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Timestamp;
+import java.util.List;
+
+public class TaskStatusRepository extends ExpCatAbstractRepository<TaskStatus, TaskStatusEntity, TaskStatusPK> {
+ private final static Logger logger = LoggerFactory.getLogger(TaskStatusRepository.class);
+
+ public TaskStatusRepository() { super(TaskStatus.class, TaskStatusEntity.class); }
+
+ protected String saveTaskStatus(TaskStatus taskStatus, String taskId) throws RegistryException {
+ Mapper mapper = ObjectMapperSingleton.getInstance();
+ TaskStatusEntity taskStatusEntity = mapper.map(taskStatus, TaskStatusEntity.class);
+
+ if (taskStatusEntity.getTaskId() == null) {
+ logger.debug("Setting the TaskStatusEntity's TaskId");
+ taskStatusEntity.setTaskId(taskId);
+ }
+
+ execute(entityManager -> entityManager.merge(taskStatusEntity));
+ return taskStatusEntity.getStatusId();
+ }
+
+ public String addTaskStatus(TaskStatus taskStatus, String taskId) throws RegistryException {
+
+ if (taskStatus.getStatusId() == null) {
+ logger.debug("Setting the TaskStatus's StatusId");
+ taskStatus.setStatusId(ExpCatalogUtils.getID("STATUS"));
+ }
+
+ return saveTaskStatus(taskStatus, taskId);
+ }
+
+ public String updateTaskStatus(TaskStatus updatedTaskStatus, String taskId) throws RegistryException {
+ return saveTaskStatus(updatedTaskStatus, taskId);
+ }
+
+ public TaskStatus getTaskStatus(String taskId) throws RegistryException {
+ TaskRepository taskRepository = new TaskRepository();
+ TaskModel taskModel = taskRepository.getTask(taskId);
+ List<TaskStatus> taskStatusList = taskModel.getTaskStatuses();
+
+ if(taskStatusList.size() == 0) {
+ logger.debug("TaskStatus list is empty");
+ return null;
+ }
+
+ else {
+ TaskStatus latestTaskStatus = taskStatusList.get(0);
+
+ for(int i = 1; i < taskStatusList.size(); i++) {
+ Timestamp timeOfStateChange = new Timestamp(taskStatusList.get(i).getTimeOfStateChange());
+
+ if (timeOfStateChange.after(new Timestamp(latestTaskStatus.getTimeOfStateChange()))
+ || (timeOfStateChange.equals(latestTaskStatus.getTimeOfStateChange()) && taskStatusList.get(i).getState().equals(TaskState.COMPLETED.toString()))
+ || (timeOfStateChange.equals(latestTaskStatus.getTimeOfStateChange()) && taskStatusList.get(i).getState().equals(TaskState.FAILED.toString()))
+ || (timeOfStateChange.equals(latestTaskStatus.getTimeOfStateChange()) && taskStatusList.get(i).getState().equals(TaskState.CANCELED.toString()))) {
+ latestTaskStatus = taskStatusList.get(i);
+ }
+
+ }
+
+ return latestTaskStatus;
+ }
+ }
+
+}

ADD
modules/registry-refactoring/src/test/java/org/apache/airavata/registry/core/repositories/expcatalog/TaskErrorRepositoryTest.java
+package org.apache.airavata.registry.core.repositories.expcatalog;
+
+import org.apache.airavata.model.commons.ErrorModel;
+import org.apache.airavata.model.experiment.ExperimentModel;
+import org.apache.airavata.model.experiment.ExperimentType;
+import org.apache.airavata.model.process.ProcessModel;
+import org.apache.airavata.model.task.TaskModel;
+import org.apache.airavata.model.task.TaskTypes;
+import org.apache.airavata.model.workspace.Gateway;
+import org.apache.airavata.model.workspace.Project;
+import org.apache.airavata.registry.core.repositories.expcatalog.util.Initialize;
+import org.apache.airavata.registry.cpi.RegistryException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TaskErrorRepositoryTest {
+
+ private static Initialize initialize;
+ GatewayRepository gatewayRepository;
+ ProjectRepository projectRepository;
+ ExperimentRepository experimentRepository;
+ ProcessRepository processRepository;
+ TaskRepository taskRepository;
+ TaskErrorRepository taskErrorRepository;
+ private static final Logger logger = LoggerFactory.getLogger(TaskErrorRepositoryTest.class);
+
+ @Before
+ public void setUp() {
+ try {
+ initialize = new Initialize("expcatalog-derby.sql");
+ initialize.initializeDB();
+ gatewayRepository = new GatewayRepository();
+ projectRepository = new ProjectRepository();
+ experimentRepository = new ExperimentRepository();
+ processRepository = new ProcessRepository();
+ taskRepository = new TaskRepository();
+ taskErrorRepository = new TaskErrorRepository();
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ System.out.println("********** TEAR DOWN ************");
+ initialize.stopDerbyServer();
+ }
+
+ @Test
+ public void TaskErrorRepositoryTest() throws RegistryException {
+ Gateway gateway = new Gateway();
+ gateway.setGatewayId("gateway");
+ gateway.setDomain("SEAGRID");
+ gateway.setEmailAddress("abc@d.com");
+ String gatewayId = gatewayRepository.addGateway(gateway);
+
+ Project project = new Project();
+ project.setName("projectName");
+ project.setOwner("user");
+ project.setGatewayId(gatewayId);
+
+ String projectId = projectRepository.addProject(project, gatewayId);
+
+ ExperimentModel experimentModel = new ExperimentModel();
+ experimentModel.setProjectId(projectId);
+ experimentModel.setGatewayId(gatewayId);
+ experimentModel.setExperimentType(ExperimentType.SINGLE_APPLICATION);
+ experimentModel.setUserName("user");
+ experimentModel.setExperimentName("name");
+
+ String experimentId = experimentRepository.addExperiment(experimentModel);
+
+ ProcessModel processModel = new ProcessModel(null, experimentId);
+ String processId = processRepository.addProcess(processModel, experimentId);
+
+ TaskModel taskModel = new TaskModel();
+ taskModel.setTaskType(TaskTypes.JOB_SUBMISSION);
+ taskModel.setParentProcessId(processId);
+
+ String taskId = taskRepository.addTask(taskModel, processId);
+ assertTrue(taskId != null);
+
+ ErrorModel errorModel = new ErrorModel();
+ errorModel.setErrorId("error");
+
+ String taskErrorId = taskErrorRepository.addTaskError(errorModel, taskId);
+ assertTrue(taskErrorId != null);
+ assertTrue(taskRepository.getTask(taskId).getTaskErrors().size() == 1);
+
+ errorModel.setActualErrorMessage("message");
+ taskErrorRepository.updateTaskError(errorModel, taskId);
+
+ List<ErrorModel> retrievedErrorList = taskErrorRepository.getTaskError(taskId);
+ assertTrue(retrievedErrorList.size() == 1);
+ assertEquals("message", retrievedErrorList.get(0).getActualErrorMessage());
+
+ experimentRepository.removeExperiment(experimentId);
+ processRepository.removeProcess(processId);
+ taskRepository.removeTask(taskId);
+ gatewayRepository.removeGateway(gatewayId);
+ projectRepository.removeProject(projectId);
+ }
+
+}

MODIFY
modules/registry-refactoring/src/test/java/org/apache/airavata/registry/core/repositories/expcatalog/TaskRepositoryTest.java
modules/registry-refactoring/src/test/java/org/apache/airavata/registry/core/repositories/expcatalog/TaskRepositoryTest.java
*/
package org.apache.airavata.registry.core.repositories.expcatalog;
-import org.apache.airavata.model.commons.ErrorModel;
import org.apache.airavata.model.experiment.ExperimentModel;
import org.apache.airavata.model.experiment.ExperimentType;
import org.apache.airavata.model.process.ProcessModel;
-import org.apache.airavata.model.status.TaskState;
-import org.apache.airavata.model.status.TaskStatus;
import org.apache.airavata.model.task.TaskModel;
import org.apache.airavata.model.task.TaskTypes;
import org.apache.airavata.model.workspace.Gateway;
@@ -114,28 +111,6 @@ public class TaskRepositoryTest {
taskRepository.updateTask(taskModel, taskId);
assertEquals(TaskTypes.MONITORING, taskRepository.getTask(taskId).getTaskType());
- TaskStatus taskStatus = new TaskStatus(TaskState.EXECUTING);
- String taskStatusId = taskRepository.addTaskStatus(taskStatus, taskId);
- assertTrue(taskStatusId != null);
-
- taskStatus.setState(TaskState.CREATED);
- taskRepository.updateTaskStatus(taskStatus, taskId);
-
- TaskStatus retrievedTaskStatus = taskRepository.getTaskStatus(taskId);
- assertEquals(TaskState.CREATED, retrievedTaskStatus.getState());
-
- ErrorModel errorModel = new ErrorModel();
- errorModel.setErrorId("error");
-
- assertEquals(taskId, taskRepository.addTaskError(errorModel, taskId));
-
- errorModel.setActualErrorMessage("message");
- taskRepository.updateTaskError(errorModel, taskId);
-
- List<ErrorModel> retrievedErrorList = taskRepository.getTaskError(taskId);
- assertTrue(retrievedErrorList.size() == 1);
- assertEquals("message", retrievedErrorList.get(0).getActualErrorMessage());
-
List<String> taskIdList = taskRepository.getTaskIds(DBConstants.Task.PARENT_PROCESS_ID, processId);
assertTrue(taskIdList.size() == 1);
assertTrue(taskIdList.get(0).equals(taskId));

ADD
modules/registry-refactoring/src/test/java/org/apache/airavata/registry/core/repositories/expcatalog/TaskStatusRepositoryTest.java
+package org.apache.airavata.registry.core.repositories.expcatalog;
+
+import org.apache.airavata.model.experiment.ExperimentModel;
+import org.apache.airavata.model.experiment.ExperimentType;
+import org.apache.airavata.model.process.ProcessModel;
+import org.apache.airavata.model.status.TaskState;
+import org.apache.airavata.model.status.TaskStatus;
+import org.apache.airavata.model.task.TaskModel;
+import org.apache.airavata.model.task.TaskTypes;
+import org.apache.airavata.model.workspace.Gateway;
+import org.apache.airavata.model.workspace.Project;
+import org.apache.airavata.registry.core.repositories.expcatalog.util.Initialize;
+import org.apache.airavata.registry.cpi.RegistryException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TaskStatusRepositoryTest {
+
+ private static Initialize initialize;
+ GatewayRepository gatewayRepository;
+ ProjectRepository projectRepository;
+ ExperimentRepository experimentRepository;
+ ProcessRepository processRepository;
+ TaskRepository taskRepository;
+ TaskStatusRepository taskStatusRepository;
+ private static final Logger logger = LoggerFactory.getLogger(TaskStatusRepositoryTest.class);
+
+ @Before
+ public void setUp() {
+ try {
+ initialize = new Initialize("expcatalog-derby.sql");
+ initialize.initializeDB();
+ gatewayRepository = new GatewayRepository();
+ projectRepository = new ProjectRepository();
+ experimentRepository = new ExperimentRepository();
+ processRepository = new ProcessRepository();
+ taskRepository = new TaskRepository();
+ taskStatusRepository = new TaskStatusRepository();
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ System.out.println("********** TEAR DOWN ************");
+ initialize.stopDerbyServer();
+ }
+
+ @Test
+ public void TaskStatusRepositoryTest() throws RegistryException {
+ Gateway gateway = new Gateway();
+ gateway.setGatewayId("gateway");
+ gateway.setDomain("SEAGRID");
+ gateway.setEmailAddress("abc@d.com");
+ String gatewayId = gatewayRepository.addGateway(gateway);
+
+ Project project = new Project();
+ project.setName("projectName");
+ project.setOwner("user");
+ project.setGatewayId(gatewayId);
+
+ String projectId = projectRepository.addProject(project, gatewayId);
+
+ ExperimentModel experimentModel = new ExperimentModel();
+ experimentModel.setProjectId(projectId);
+ experimentModel.setGatewayId(gatewayId);
+ experimentModel.setExperimentType(ExperimentType.SINGLE_APPLICATION);
+ experimentModel.setUserName("user");
+ experimentModel.setExperimentName("name");
+
+ String experimentId = experimentRepository.addExperiment(experimentModel);
+
+ ProcessModel processModel = new ProcessModel(null, experimentId);
+ String processId = processRepository.addProcess(processModel, experimentId);
+
+ TaskModel taskModel = new TaskModel();
+ taskModel.setTaskType(TaskTypes.JOB_SUBMISSION);
+ taskModel.setParentProcessId(processId);
+
+ String taskId = taskRepository.addTask(taskModel, processId);
+ assertTrue(taskId != null);
+
+ TaskStatus taskStatus = new TaskStatus(TaskState.EXECUTING);
+ String taskStatusId = taskStatusRepository.addTaskStatus(taskStatus, taskId);
+ assertTrue(taskStatusId != null);
+ assertTrue(taskRepository.getTask(taskId).getTaskStatuses().size() == 1);
+
+ taskStatus.setState(TaskState.CREATED);
+ taskStatusRepository.updateTaskStatus(taskStatus, taskId);
+
+ TaskStatus retrievedTaskStatus = taskStatusRepository.getTaskStatus(taskId);
+ assertEquals(TaskState.CREATED, retrievedTaskStatus.getState());
+
+ experimentRepository.removeExperiment(experimentId);
+ processRepository.removeProcess(processId);
+ taskRepository.removeTask(taskId);
+ gatewayRepository.removeGateway(gatewayId);
+ projectRepository.removeProject(projectId);
+ }
+
+}

MODIFY
modules/registry/registry-server/registry-api-service/src/main/java/org/apache/airavata/registry/api/service/handler/RegistryServerHandler.java
modules/registry/registry-server/registry-api-service/src/main/java/org/apache/airavata/registry/api/service/handler/RegistryServerHandler.java
@@ -152,6 +152,8 @@ public class RegistryServerHandler implements RegistryService.Iface {
private ProcessStatusRepository processStatusRepository = new ProcessStatusRepository();
private ProcessErrorRepository processErrorRepository = new ProcessErrorRepository();
private TaskRepository taskRepository = new TaskRepository();
+ private TaskStatusRepository taskStatusRepository = new TaskStatusRepository();
+ private TaskErrorRepository taskErrorRepository = new TaskErrorRepository();
private JobRepository jobRepository = new JobRepository();
private QueueStatusRepository queueStatusRepository = new QueueStatusRepository();
private DataProductRepository dataProductRepository = new DataProductRepository();
@@ -819,7 +821,7 @@ public class RegistryServerHandler implements RegistryService.Iface {
experimentErrorRepository.addExperimentError(errorModel, id);
}
else if (ExpCatChildDataType.TASK_ERROR.equals(ExpCatChildDataType.valueOf(errorType))) {
- taskRepository.addTaskError(errorModel, id);
+ taskErrorRepository.addTaskError(errorModel, id);
}
else if (ExpCatChildDataType.PROCESS_ERROR.equals(ExpCatChildDataType.valueOf(errorType))) {
processErrorRepository.addProcessError(errorModel, id);
@@ -836,7 +838,7 @@ public class RegistryServerHandler implements RegistryService.Iface {
@Override
public void addTaskStatus(TaskStatus taskStatus, String taskId) throws RegistryServiceException, TException {
try {
- taskRepository.addTaskStatus(taskStatus, taskId);
+ taskStatusRepository.addTaskStatus(taskStatus, taskId);
} catch (Exception e) {
logger.error(taskId, "Error while adding task status", e);
AiravataSystemException exception = new AiravataSystemException();

Adding new task repositories