Merge pull request #18495 from mvdbeek/dont_call_stop_job_for_new_jobs

[24.1] Don't call job_runner.stop_job on jobs in new state

lib/galaxy/jobs/handler.py

@@ -1262,7 +1262,8 @@ def stop(self, job, job_wrapper):
             runner_name = job_runner_name.split(":", 1)[0]
             log.debug(f"Stopping job {job_wrapper.get_id_tag()} in {runner_name} runner")
             try:
-                self.job_runners[runner_name].stop_job(job_wrapper)
+                if job.state != model.Job.states.NEW:
+                    self.job_runners[runner_name].stop_job(job_wrapper)
             except KeyError:
                 log.error(f"stop(): ({job_wrapper.get_id_tag()}) Invalid job runner: {runner_name}")
                 # Job and output dataset states have already been updated, so nothing is done here.

lib/galaxy/jobs/runners/__init__.py

@@ -581,7 +581,8 @@ def _handle_runner_state(self, runner_state, job_state: "JobState"):
             log.exception("Caught exception in runner state handler")
 
     def fail_job(self, job_state: "JobState", exception=False, message="Job failed", full_status=None):
-        if getattr(job_state, "stop_job", True):
+        job = job_state.job_wrapper.get_job()
+        if getattr(job_state, "stop_job", True) and job.state != model.Job.states.NEW:
             self.stop_job(job_state.job_wrapper)
         job_state.job_wrapper.reclaim_ownership()
         self._handle_runner_state("failure", job_state)

lib/galaxy/jobs/runners/aws.py

@@ -418,7 +418,8 @@ def recover(self, job, job_wrapper):
             self.monitor_queue.put(ajs)
 
     def fail_job(self, job_state, exception=False):
-        if getattr(job_state, "stop_job", True):
+        job = job_state.job_wrapper.get_job()
+        if getattr(job_state, "stop_job", True) and job.state != model.Job.states.NEW:
             self.stop_job(job_state.job_wrapper)
         job_state.job_wrapper.reclaim_ownership()
         self._handle_runner_state("failure", job_state)