vincbeck commented on code in PR #75907:
URL: https://github.com/apache/airflow/pull/75907#discussion_r4413270078
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/ecs/ecs_executor.py:
##########
@@ -570,17 +627,27 @@ def end(self, heartbeat_interval=10):
try:
while True:
self.sync()
- if not self.active_workers:
+ if not self.active_workers and not self._run_task_futures:
break
time.sleep(heartbeat_interval)
except Exception:
# We catch any and all exceptions because otherwise they would
bubble
# up and kill the scheduler process.
self.log.exception("Failed to end %s", self.__class__.__name__)
+ if self._run_task_pool is not None:
+ self._run_task_pool.shutdown(wait=False, cancel_futures=True)
def terminate(self):
"""Kill all ECS processes by calling Boto3's StopTask API."""
try:
+ if self._run_task_pool is not None:
+ # A RunTask call still in flight may start an ECS task after
the StopTask calls below:
+ # wait for the calls already started, and drop the ones not
started yet.
+ self._run_task_pool.shutdown(wait=True, cancel_futures=True)
Review Comment:
`terminate()` sets `_run_task_pool = None` after the shutdown, so `end()`
makes its calls in the scheduler loop, as on main
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
