Throw GenieConflictException if a job request comes in with an id tha…

…t is already running. (#437) * Throw GenieConflictException if a job request comes in with an id that is already running. * Throw GenieConflictException if a job request comes in with an id that is already running. * Throw GenieConflictException if a job request comes in with an id that is already running.
Netflix · Nov 17, 2016 · 7f582b9 · 7f582b9
1 parent 03827c3
commit 7f582b9
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 6 deletions.
diff --git a/genie-core/src/main/java/com/netflix/genie/core/services/impl/JobCoordinatorServiceImpl.java b/genie-core/src/main/java/com/netflix/genie/core/services/impl/JobCoordinatorServiceImpl.java
@@ -26,6 +26,7 @@
 import com.netflix.genie.common.dto.JobMetadata;
 import com.netflix.genie.common.dto.JobRequest;
 import com.netflix.genie.common.dto.JobStatus;
+import com.netflix.genie.common.exceptions.GenieConflictException;
 import com.netflix.genie.common.exceptions.GenieException;
 import com.netflix.genie.common.exceptions.GeniePreconditionException;
 import com.netflix.genie.common.exceptions.GenieServerException;
@@ -154,7 +155,6 @@ public String coordinateJob(
         JobStatus jobStatus = JobStatus.FAILED;
         try {
             log.info("Called to schedule job launch for job {}", jobId);
-            jobStateService.init(jobId);
             // create the job object in the database with status INIT
             final Job.Builder jobBuilder = new Job.Builder(
                 jobRequest.getName(),
@@ -183,7 +183,7 @@ public String coordinateJob(
 
             // Log all the job initial job information
             this.jobPersistenceService.createJob(jobRequest, jobMetadata, jobBuilder.build(), jobExecution);
-
+            jobStateService.init(jobId);
             //TODO: Combine the cluster and command selection into a single method/database query for efficiency
             // Resolve the cluster for the job request based on the tags specified
             final Cluster cluster = this.getCluster(jobRequest);
@@ -240,13 +240,28 @@ public String coordinateJob(
                     );
                 }
             }
+        } catch (GenieConflictException e) {
+            // Job has not been initiated so we don't have to call JobStateService.done()
+            throw e;
         } catch (GenieException e) {
-            jobStateService.done(jobId);
-            jobPersistenceService.updateJobStatus(jobId, jobStatus, e.getMessage());
+            //
+            // Need to check if the job exists in the JobStateService
+            // because this error can happen before the job is initiated.
+            //
+            if (jobStateService.jobExists(jobId)) {
+                jobStateService.done(jobId);
+                jobPersistenceService.updateJobStatus(jobId, jobStatus, e.getMessage());
+            }
             throw e;
         } catch (Exception e) {
-            jobStateService.done(jobId);
-            jobPersistenceService.updateJobStatus(jobId, jobStatus, e.getMessage());
+            //
+            // Need to check if the job exists in the JobStateService
+            // because this error can happen before the job is initiated.
+            //
+            if (jobStateService.jobExists(jobId)) {
+                jobStateService.done(jobId);
+                jobPersistenceService.updateJobStatus(jobId, jobStatus, e.getMessage());
+            }
             throw new GenieServerException(e);
         } finally {
             this.coordinationTimer.record(System.nanoTime() - coordinationStart, TimeUnit.MILLISECONDS);

diff --git a/...rc/test/java/com/netflix/genie/core/services/impl/JobCoordinatorServiceImplUnitTests.java b/...rc/test/java/com/netflix/genie/core/services/impl/JobCoordinatorServiceImplUnitTests.java
@@ -28,6 +28,7 @@
 import com.netflix.genie.common.dto.JobMetadata;
 import com.netflix.genie.common.dto.JobRequest;
 import com.netflix.genie.common.dto.JobStatus;
+import com.netflix.genie.common.exceptions.GenieConflictException;
 import com.netflix.genie.common.exceptions.GenieException;
 import com.netflix.genie.common.exceptions.GeniePreconditionException;
 import com.netflix.genie.common.exceptions.GenieServerException;
@@ -478,6 +479,22 @@ public void cantCoordinateIfNoId() throws GenieException {
         this.jobCoordinatorService.coordinateJob(request, Mockito.mock(JobMetadata.class));
     }
 
+    /**
+     * Make sure if the job with id already exists.
+     *
+     * @throws GenieException On error
+     */
+    @Test(expected = GenieConflictException.class)
+    public void cantCoordinateIfJobAlreadyExists() throws GenieException {
+        final JobRequest request = getJobRequest(false, null, null, null);
+        final JobMetadata metadata = Mockito.mock(JobMetadata.class);
+        Mockito.doThrow(GenieConflictException.class).when(jobPersistenceService)
+            .createJob(Mockito.eq(request), Mockito.eq(metadata),
+                Mockito.any(Job.class),
+                Mockito.any(JobExecution.class));
+        this.jobCoordinatorService.coordinateJob(request, metadata);
+    }
+
     /**
      * Test killing a job without throwing an exception.
      *