Force updating the job status to KILLED when killing a job that has a…

… connected agent but no response observer
Netflix · Oct 20, 2023 · ce86e8c · ce86e8c
1 parent c6c81dc
commit ce86e8c
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 10 deletions.
diff --git a/...c/main/java/com/netflix/genie/web/agent/apis/rpc/v4/endpoints/GRpcJobKillServiceImpl.java b/...c/main/java/com/netflix/genie/web/agent/apis/rpc/v4/endpoints/GRpcJobKillServiceImpl.java
@@ -154,15 +154,35 @@ public void killJob(
                     = this.parkedJobKillResponseObservers.remove(jobId);
 
                 if (responseObserver == null) {
-                    log.error("Job {} not killed. Expected local agent connection not found", jobId);
-                    throw new GenieServerException(
-                        "Job " + jobId + " not killed. Expected local agent connection not found."
+                    // This might happen when the agent has gone but its status is not updated
+                    // In this case, we force updating the job status to KILLED.
+                    log.warn("Job {} not killed. Expected local agent connection not found. "
+                            + "Trying to force updating the job status to {}",
+                        jobId,
+                        JobStatus.KILLED
                     );
-                }
-                responseObserver.onNext(JobKillRegistrationResponse.newBuilder().build());
-                responseObserver.onCompleted();
+                    try {
+                        this.persistenceService.updateJobStatus(jobId, currentJobStatus, JobStatus.KILLED, reason);
+                        log.info("Succeeded to force updating the status of Job {} to {}",
+                            jobId,
+                            JobStatus.KILLED
+                        );
+                    } catch (final Exception e) {
+                        log.error("Succeeded to force updating the status of Job {} to {}",
+                            jobId,
+                            JobStatus.KILLED
+                        );
+                        throw new GenieServerException("Failed to force updating the status of Job "
+                            + jobId + " to " + JobStatus.KILLED,
+                            e
+                        );
+                    }
+                } else {
+                    responseObserver.onNext(JobKillRegistrationResponse.newBuilder().build());
+                    responseObserver.onCompleted();
 
-                log.info("Agent notified for killing job {}", jobId);
+                    log.info("Agent notified for killing job {}", jobId);
+                }
             } else {
                 // Agent is running somewhere else try to forward the request
                 final String hostname = this.agentRoutingService

diff --git a/...roovy/com/netflix/genie/web/agent/apis/rpc/v4/endpoints/GRpcJobKillServiceImplSpec.groovy b/...roovy/com/netflix/genie/web/agent/apis/rpc/v4/endpoints/GRpcJobKillServiceImplSpec.groovy
@@ -148,13 +148,12 @@ class GRpcJobKillServiceImplSpec extends Specification {
         when: "The job is active, the agent is connected, the job is local but no observer"
         this.serviceSpy.killJob(this.jobId, this.reason, this.servletRequest)
 
-        then: "Correct exception is thrown"
+        then: "Force updating job status"
         1 * this.persistenceService.getJobStatus(this.jobId) >> JobStatus.CLAIMED
-        0 * this.persistenceService.updateJobStatus(_ as String, _ as JobStatus, _ as JobStatus, _ as String)
+        1 * this.persistenceService.updateJobStatus(_ as String, _ as JobStatus, _ as JobStatus, _ as String)
         1 * this.agentRoutingService.isAgentConnectionLocal(this.jobId) >> true
         0 * this.responseObserver.onNext(_ as JobKillRegistrationResponse)
         0 * this.responseObserver.onCompleted()
-        thrown(GenieServerException)
 
         when: "The job is active, the agent is connected, and there is an observer"
         this.serviceSpy.registerForKillNotification(this.request, this.responseObserver)