Adding new job repository
MODIFY
modules/registry-refactoring/src/main/java/org/apache/airavata/registry/core/repositories/expcatalog/JobRepository.java
modules/registry-refactoring/src/main/java/org/apache/airavata/registry/core/repositories/expcatalog/JobRepository.java
@@ -22,11 +22,9 @@ package org.apache.airavata.registry.core.repositories.expcatalog;
import org.apache.airavata.model.commons.airavata_commonsConstants;
import org.apache.airavata.model.job.JobModel;
-import org.apache.airavata.model.status.JobStatus;
import org.apache.airavata.registry.core.entities.expcatalog.JobEntity;
import org.apache.airavata.registry.core.entities.expcatalog.JobPK;
import org.apache.airavata.registry.core.utils.DBConstants;
-import org.apache.airavata.registry.core.utils.ExpCatalogUtils;
import org.apache.airavata.registry.core.utils.ObjectMapperSingleton;
import org.apache.airavata.registry.core.utils.QueryConstants;
import org.apache.airavata.registry.cpi.RegistryException;
@@ -93,52 +91,6 @@ public class JobRepository extends ExpCatAbstractRepository<JobModel, JobEntity,
return get(jobPK);
}
- public String addJobStatus(JobStatus jobStatus, JobPK jobPK) throws RegistryException {
- JobModel jobModel = getJob(jobPK);
- List<JobStatus> jobStatusList = jobModel.getJobStatuses();
-
- if (jobStatusList.size() == 0 || !jobStatusList.contains(jobStatus)) {
-
- if (jobStatus.getStatusId() == null) {
- logger.debug("Set JobStatus's StatusId");
- jobStatus.setStatusId(ExpCatalogUtils.getID("STATUS"));
- }
-
- logger.debug("Adding the JobStatus to the list");
- jobStatusList.add(jobStatus);
-
- }
-
- jobModel.setJobStatuses(jobStatusList);
- updateJob(jobModel, jobPK);
- return jobStatus.getStatusId();
- }
-
- public String updateJobStatus(JobStatus updatedJobStatus, JobPK jobPK) throws RegistryException {
- JobModel jobModel = getJob(jobPK);
- List<JobStatus> jobStatusList = jobModel.getJobStatuses();
-
- for (JobStatus retrievedJobStatus : jobStatusList) {
-
- if (retrievedJobStatus.getStatusId().equals(updatedJobStatus.getStatusId())) {
- logger.debug("Updating the JobStatus");
- jobStatusList.remove(retrievedJobStatus);
- jobStatusList.add(updatedJobStatus);
- }
-
- }
-
- jobModel.setJobStatuses(jobStatusList);
- updateJob(jobModel, jobPK);
- return updatedJobStatus.getStatusId();
-
- }
-
- public List<JobStatus> getJobStatus(JobPK jobPK) throws RegistryException {
- JobModel jobModel = getJob(jobPK);
- return jobModel.getJobStatuses();
- }
-
public List<JobModel> getJobList(String fieldName, Object value) throws RegistryException {
JobRepository jobRepository = new JobRepository();
List<JobModel> jobModelList;

ADD
modules/registry-refactoring/src/main/java/org/apache/airavata/registry/core/repositories/expcatalog/JobStatusRepository.java
+/*
+ *
+ * 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.airavata.registry.core.repositories.expcatalog;
+
+import org.apache.airavata.model.job.JobModel;
+import org.apache.airavata.model.status.JobState;
+import org.apache.airavata.model.status.JobStatus;
+import org.apache.airavata.registry.core.entities.expcatalog.JobPK;
+import org.apache.airavata.registry.core.entities.expcatalog.JobStatusEntity;
+import org.apache.airavata.registry.core.entities.expcatalog.JobStatusPK;
+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 JobStatusRepository extends ExpCatAbstractRepository<JobStatus, JobStatusEntity, JobStatusPK> {
+ private final static Logger logger = LoggerFactory.getLogger(JobStatusRepository.class);
+
+ public JobStatusRepository() { super(JobStatus.class, JobStatusEntity.class); }
+
+ protected String saveJobStatus(JobStatus jobStatus, JobPK jobPK) throws RegistryException {
+ Mapper mapper = ObjectMapperSingleton.getInstance();
+ JobStatusEntity jobStatusEntity = mapper.map(jobStatus, JobStatusEntity.class);
+
+ if (jobStatusEntity.getJobId() == null) {
+ logger.debug("Setting the JobStatusEntity's JobId");
+ jobStatusEntity.setJobId(jobPK.getJobId());
+ }
+
+ if (jobStatusEntity.getTaskId() == null) {
+ logger.debug("Setting the JobStatusEntity's TaskId");
+ jobStatusEntity.setTaskId(jobPK.getTaskId());
+ }
+
+ execute(entityManager -> entityManager.merge(jobStatusEntity));
+ return jobStatusEntity.getStatusId();
+ }
+
+ public String addJobStatus(JobStatus jobStatus, JobPK jobPK) throws RegistryException {
+
+ if (jobStatus.getStatusId() == null) {
+ logger.debug("Setting the JobStatusEntity's StatusId");
+ jobStatus.setStatusId(ExpCatalogUtils.getID("STATUS"));
+ }
+
+ return saveJobStatus(jobStatus, jobPK);
+
+ }
+
+ public String updateJobStatus(JobStatus updatedJobStatus, JobPK jobPK) throws RegistryException {
+ return saveJobStatus(updatedJobStatus, jobPK);
+ }
+
+ public JobStatus getJobStatus(JobPK jobPK) throws RegistryException {
+ JobRepository jobRepository = new JobRepository();
+ JobModel jobModel = jobRepository.getJob(jobPK);
+ List<JobStatus> jobStatusList = jobModel.getJobStatuses();
+
+ if(jobStatusList.size() == 0) {
+ logger.debug("JobStatus list is empty");
+ return null;
+ }
+
+ else {
+ JobStatus latestJobStatus = jobStatusList.get(0);
+
+ for(int i = 1; i < jobStatusList.size(); i++) {
+ Timestamp timeOfStateChange = new Timestamp(jobStatusList.get(i).getTimeOfStateChange());
+
+ if (timeOfStateChange.after(new Timestamp(latestJobStatus.getTimeOfStateChange()))
+ || (timeOfStateChange.equals(latestJobStatus.getTimeOfStateChange()) && jobStatusList.get(i).getJobState().equals(JobState.COMPLETE.toString()))
+ || (timeOfStateChange.equals(latestJobStatus.getTimeOfStateChange()) && jobStatusList.get(i).getJobState().equals(JobState.FAILED.toString()))
+ || (timeOfStateChange.equals(latestJobStatus.getTimeOfStateChange()) && jobStatusList.get(i).getJobState().equals(JobState.CANCELED.toString()))) {
+ latestJobStatus = jobStatusList.get(i);
+ }
+
+ }
+
+ return latestJobStatus;
+ }
+ }
+
+
+}

MODIFY
modules/registry-refactoring/src/test/java/org/apache/airavata/registry/core/repositories/expcatalog/JobRepositoryTest.java
modules/registry-refactoring/src/test/java/org/apache/airavata/registry/core/repositories/expcatalog/JobRepositoryTest.java
@@ -24,8 +24,6 @@ import org.apache.airavata.model.experiment.ExperimentModel;
import org.apache.airavata.model.experiment.ExperimentType;
import org.apache.airavata.model.job.JobModel;
import org.apache.airavata.model.process.ProcessModel;
-import org.apache.airavata.model.status.JobState;
-import org.apache.airavata.model.status.JobStatus;
import org.apache.airavata.model.task.TaskModel;
import org.apache.airavata.model.task.TaskTypes;
import org.apache.airavata.model.workspace.Gateway;
@@ -132,17 +130,6 @@ public class JobRepositoryTest {
jobRepository.updateJob(jobModel, jobPK);
assertEquals("jobName", jobRepository.getJob(jobPK).getJobName());
- JobStatus jobStatus = new JobStatus(JobState.QUEUED);
- String jobStatusId = jobRepository.addJobStatus(jobStatus, jobPK);
- assertTrue(jobStatusId != null);
-
- jobStatus.setJobState(JobState.ACTIVE);
- jobRepository.updateJobStatus(jobStatus, jobPK);
-
- List<JobStatus> retrievedJobStatusList = jobRepository.getJobStatus(jobPK);
- assertTrue(retrievedJobStatusList.size() == 1);
- assertEquals(JobState.ACTIVE, retrievedJobStatusList.get(0).getJobState());
-
List<String> jobIdList = jobRepository.getJobIds(DBConstants.Job.TASK_ID, taskId);
assertTrue(jobIdList.size() == 1);
assertTrue(jobIdList.get(0).equals(jobId));

ADD
modules/registry-refactoring/src/test/java/org/apache/airavata/registry/core/repositories/expcatalog/JobStatusRepositoryTest.java
+/*
+ *
+ * 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.airavata.registry.core.repositories.expcatalog;
+
+import org.apache.airavata.model.experiment.ExperimentModel;
+import org.apache.airavata.model.experiment.ExperimentType;
+import org.apache.airavata.model.job.JobModel;
+import org.apache.airavata.model.process.ProcessModel;
+import org.apache.airavata.model.status.JobState;
+import org.apache.airavata.model.status.JobStatus;
+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.entities.expcatalog.JobPK;
+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 JobStatusRepositoryTest {
+
+ private static Initialize initialize;
+ GatewayRepository gatewayRepository;
+ ProjectRepository projectRepository;
+ ExperimentRepository experimentRepository;
+ ProcessRepository processRepository;
+ TaskRepository taskRepository;
+ JobRepository jobRepository;
+ JobStatusRepository jobStatusRepository;
+ private static final Logger logger = LoggerFactory.getLogger(JobStatusRepositoryTest.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();
+ jobRepository = new JobRepository();
+ jobStatusRepository = new JobStatusRepository();
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ System.out.println("********** TEAR DOWN ************");
+ initialize.stopDerbyServer();
+ }
+
+ @Test
+ public void JobStatusRepositoryTest() 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);
+
+ taskModel.setTaskType(TaskTypes.MONITORING);
+ taskRepository.updateTask(taskModel, taskId);
+
+ JobPK jobPK = new JobPK();
+ jobPK.setJobId("job");
+ jobPK.setTaskId(taskId);
+
+ JobModel jobModel = new JobModel();
+ jobModel.setJobId(jobPK.getJobId());
+ jobModel.setTaskId(jobPK.getTaskId());
+ jobModel.setJobDescription("jobDescription");
+
+ String jobId = jobRepository.addJob(jobModel, processId);
+ assertTrue(jobId != null);
+
+ JobStatus jobStatus = new JobStatus(JobState.QUEUED);
+ String jobStatusId = jobStatusRepository.addJobStatus(jobStatus, jobPK);
+ assertTrue(jobStatusId != null);
+ assertTrue(jobRepository.getJob(jobPK).getJobStatuses().size() == 1);
+
+ jobStatus.setJobState(JobState.ACTIVE);
+ jobStatusRepository.updateJobStatus(jobStatus, jobPK);
+
+ JobStatus retrievedJobStatus = jobStatusRepository.getJobStatus(jobPK);
+ assertEquals(JobState.ACTIVE, retrievedJobStatus.getJobState());
+
+ experimentRepository.removeExperiment(experimentId);
+ processRepository.removeProcess(processId);
+ taskRepository.removeTask(taskId);
+ jobRepository.removeJob(jobPK);
+ 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
@@ -112,10 +112,8 @@ import org.apache.airavata.registry.core.repositories.replicacatalog.DataReplica
import org.apache.airavata.registry.core.repositories.workflowcatalog.WorkflowRepository;
import org.apache.airavata.registry.core.utils.DBConstants;
import org.apache.airavata.registry.cpi.AppCatalogException;
-import org.apache.airavata.registry.cpi.CompositeIdentifier;
import org.apache.airavata.registry.cpi.ComputeResource;
import org.apache.airavata.registry.cpi.ExpCatChildDataType;
-import org.apache.airavata.registry.cpi.ExperimentCatalog;
import org.apache.airavata.registry.cpi.ExperimentCatalogException;
import org.apache.airavata.registry.cpi.RegistryException;
import org.apache.airavata.registry.cpi.ResultOrderType;
@@ -155,6 +153,7 @@ public class RegistryServerHandler implements RegistryService.Iface {
private TaskStatusRepository taskStatusRepository = new TaskStatusRepository();
private TaskErrorRepository taskErrorRepository = new TaskErrorRepository();
private JobRepository jobRepository = new JobRepository();
+ private JobStatusRepository jobStatusRepository = new JobStatusRepository();
private QueueStatusRepository queueStatusRepository = new QueueStatusRepository();
private DataProductRepository dataProductRepository = new DataProductRepository();
private DataReplicaLocationRepository dataReplicaLocationRepository = new DataReplicaLocationRepository();
@@ -893,7 +892,7 @@ public class RegistryServerHandler implements RegistryService.Iface {
JobPK jobPK = new JobPK();
jobPK.setJobId(jobId);
jobPK.setTaskId(taskId);
- jobRepository.addJobStatus(jobStatus, jobPK);
+ jobStatusRepository.addJobStatus(jobStatus, jobPK);
} catch (Exception e) {
logger.error(jobId, "Error while adding job status", e);
AiravataSystemException exception = new AiravataSystemException();

Adding new job repository