return False when run launchers try to terminate a run that is alread…

…y finished

Summary:
Make this is a no-op rather than moving a run from a terminal state into a non-terminal state.

python_modules/dagster/dagster/_core/launcher/default_run_launcher.py

@@ -157,7 +157,7 @@ def terminate(self, run_id):
  return False
 
  run = self._instance.get_run_by_id(run_id)
- if not run:
+ if not run or run.is_finished:
  return False
 
  self._instance.report_run_canceling(run)

python_modules/dagster/dagster_tests/launcher_tests/test_default_run_launcher.py

@@ -473,6 +473,9 @@ def test_terminated_run(
  terminated_run = instance.get_run_by_id(run_id)
  assert terminated_run and terminated_run.status == DagsterRunStatus.CANCELED
 
+ # termination is a no-op once run is finished
+ assert not launcher.terminate(run_id)
+
  poll_for_event(
  instance,
  run_id,

python_modules/libraries/dagster-aws/dagster_aws/ecs/launcher.py

@@ -517,7 +517,7 @@ def terminate(self, run_id):
  tags = self._get_run_tags(run_id)
 
  run = self._instance.get_run_by_id(run_id)
- if not run:
+ if not run or run.is_finished:
  return False
 
  self._instance.report_run_canceling(run)

python_modules/libraries/dagster-celery-k8s/dagster_celery_k8s/launcher.py

@@ -276,7 +276,7 @@ def terminate(self, run_id):
  check.str_param(run_id, "run_id")
 
  run = self._instance.get_run_by_id(run_id)
- if not run:
+ if not run or run.is_finished:
  return False
 
  self._instance.report_run_canceling(run)

python_modules/libraries/dagster-docker/dagster_docker/docker_run_launcher.py

@@ -191,7 +191,7 @@ def _get_container(self, run):
  def terminate(self, run_id):
  run = self._instance.get_run_by_id(run_id)
 
- if not run:
+ if not run or run.is_finished:
  return False
 
  self._instance.report_run_canceling(run)

python_modules/libraries/dagster-docker/dagster_docker_tests/test_launch_docker.py

@@ -448,3 +448,6 @@ def _test_launch(
 
  poll_for_finished_run(instance, run.run_id, timeout=60)
  assert instance.get_run_by_id(run.run_id).status == DagsterRunStatus.CANCELED
+
+ # termination is a no-op once run is finished
+ assert not launcher.terminate(run.run_id)

python_modules/libraries/dagster-k8s/dagster_k8s/launcher.py

@@ -317,7 +317,7 @@ def terminate(self, run_id):
  check.str_param(run_id, "run_id")
  run = self._instance.get_run_by_id(run_id)
 
- if not run:
+ if not run or run.is_finished:
  return False
 
  self._instance.report_run_canceling(run)