vincbeck commented on code in PR #75907:
URL: https://github.com/apache/airflow/pull/75907#discussion_r4413259862
##########
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)
+
+ def _handle_run_task_response(self, ecs_task: EcsQueuedTask, get_response:
Callable[[], dict]) -> None:
+ """Get the RunTask response for ``ecs_task``, then track the ECS task,
or retry or fail the workload."""
+ 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 = []
+ try:
+ run_task_response = get_response()
+ except NoCredentialsError:
+ self.pending_workloads.append(ecs_task)
+ raise
+ except ClientError as e:
+ error_code = e.response["Error"]["Code"]
+ if error_code in INVALID_CREDENTIALS_EXCEPTIONS:
self.pending_workloads.append(ecs_task)
raise
- except ClientError as e:
- error_code = e.response["Error"]["Code"]
- if error_code in INVALID_CREDENTIALS_EXCEPTIONS:
- self.pending_workloads.append(ecs_task)
- raise
- failure_reasons.append(str(e))
- except Exception as e:
- # Failed to even get a response back from the Boto3 API or
something else went
- # wrong. For any possible failure we want to add the
exception reasons to the
- # failure list so that it is logged to the user and most
importantly the task is
- # added back to the pending list to be retried later.
- failure_reasons.append(str(e))
+ if error_code in THROTTLING_EXCEPTIONS:
Review Comment:
Added a info log statement and added backoff retries with cap
--
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]
