vincbeck commented on code in PR #75907:
URL: https://github.com/apache/airflow/pull/75907#discussion_r4413249789
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/ecs/ecs_executor.py:
##########
@@ -424,83 +463,100 @@ def attempt_workload_runs(self):
called again in the next heart-beat.
If the launch type is FARGATE, this will run the tasks on new AWS
Fargate instances.
+
+ With ``run_task_concurrency`` other than 1, the RunTask calls are made
in background threads, at
+ most that many at a time (0: no limit), and the scheduler loop does
not wait for them: each call is
+ handled by the first sync that finds it finished.
"""
- queue_len = len(self.pending_workloads)
- failure_reasons = defaultdict(int)
- for _ in range(queue_len):
+ for _ in range(len(self.pending_workloads)):
ecs_task = self.pending_workloads.popleft()
- task_key = ecs_task.key
- cmd = ecs_task.command
- queue = ecs_task.queue
- exec_config = ecs_task.executor_config
- attempt_number = ecs_task.attempt_number
- failure_reasons = []
if timezone.utcnow() < ecs_task.next_attempt_time:
self.pending_workloads.append(ecs_task)
continue
- try:
- run_task_response = self._run_task(task_key, cmd, queue,
exec_config)
- except NoCredentialsError:
+ run_task = partial(
+ self._run_task, ecs_task.key, ecs_task.command,
ecs_task.queue, ecs_task.executor_config
+ )
+ if self._run_task_pool is None:
+ self._handle_run_task_response(ecs_task, run_task)
+ else:
+ self._run_task_futures[self._run_task_pool.submit(run_task)] =
ecs_task
+ for future in [future for future in self._run_task_futures if
future.done()]:
+ self._handle_run_task_response(self._run_task_futures.pop(future),
future.result)
Review Comment:
Fixed, all finished calls are handled first, then the credentials error is
raised once
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/ecs/ecs_executor.py:
##########
@@ -424,83 +463,100 @@ def attempt_workload_runs(self):
called again in the next heart-beat.
If the launch type is FARGATE, this will run the tasks on new AWS
Fargate instances.
+
+ With ``run_task_concurrency`` other than 1, the RunTask calls are made
in background threads, at
+ most that many at a time (0: no limit), and the scheduler loop does
not wait for them: each call is
+ handled by the first sync that finds it finished.
"""
- queue_len = len(self.pending_workloads)
- failure_reasons = defaultdict(int)
- for _ in range(queue_len):
+ for _ in range(len(self.pending_workloads)):
ecs_task = self.pending_workloads.popleft()
- task_key = ecs_task.key
- cmd = ecs_task.command
- queue = ecs_task.queue
- exec_config = ecs_task.executor_config
- attempt_number = ecs_task.attempt_number
- failure_reasons = []
if timezone.utcnow() < ecs_task.next_attempt_time:
self.pending_workloads.append(ecs_task)
continue
- try:
- run_task_response = self._run_task(task_key, cmd, queue,
exec_config)
- except NoCredentialsError:
+ run_task = partial(
+ self._run_task, ecs_task.key, ecs_task.command,
ecs_task.queue, ecs_task.executor_config
+ )
+ if self._run_task_pool is None:
+ self._handle_run_task_response(ecs_task, run_task)
+ else:
+ self._run_task_futures[self._run_task_pool.submit(run_task)] =
ecs_task
+ for future in [future for future in self._run_task_futures if
future.done()]:
+ self._handle_run_task_response(self._run_task_futures.pop(future),
future.result)
Review Comment:
Fixed, all finished calls are handled first, then the credentials error is
raised once
--
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]
