o-nikolas commented on code in PR #71380:
URL: https://github.com/apache/airflow/pull/71380#discussion_r3755506447
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##########
@@ -268,11 +268,30 @@ def sync_running_jobs(self):
self.log.debug("Active Workers: %s", describe_job_response)
for job in describe_job_response:
- if job.get_job_state() == State.FAILED:
- self._handle_failed_job(job)
- elif job.get_job_state() == State.SUCCESS:
- workload_key = self.active_workers.pop_by_id(job.job_id)
- self.success(workload_key)
+ try:
+ if job.get_job_state() == State.FAILED:
+ self._handle_failed_job(job)
+ elif job.get_job_state() == State.SUCCESS:
+ workload_key = self.active_workers.pop_by_id(job.job_id)
+ self.success(workload_key)
+ except Exception:
Review Comment:
Also, if any future code in this block gets added that calls boto then
credential issues can get swallowed because it's catching Exception right now.
At the very least you could catch ClientError/CredentialsError and re-raise
those before catching Exception, but I'd still prefer to catch the exception
types we're actually worried about here.
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##########
@@ -268,11 +268,30 @@ def sync_running_jobs(self):
self.log.debug("Active Workers: %s", describe_job_response)
for job in describe_job_response:
- if job.get_job_state() == State.FAILED:
- self._handle_failed_job(job)
- elif job.get_job_state() == State.SUCCESS:
- workload_key = self.active_workers.pop_by_id(job.job_id)
- self.success(workload_key)
+ try:
+ if job.get_job_state() == State.FAILED:
+ self._handle_failed_job(job)
+ elif job.get_job_state() == State.SUCCESS:
+ workload_key = self.active_workers.pop_by_id(job.job_id)
+ self.success(workload_key)
+ except Exception:
Review Comment:
There should be more concrete exceptions we're looking for here no? Evicting
and failing tasks at any exception here feels harsh. Will every possible
exception we can receive here be constant?
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##########
@@ -268,11 +268,30 @@ def sync_running_jobs(self):
self.log.debug("Active Workers: %s", describe_job_response)
for job in describe_job_response:
- if job.get_job_state() == State.FAILED:
- self._handle_failed_job(job)
- elif job.get_job_state() == State.SUCCESS:
- workload_key = self.active_workers.pop_by_id(job.job_id)
- self.success(workload_key)
+ try:
+ if job.get_job_state() == State.FAILED:
+ self._handle_failed_job(job)
+ elif job.get_job_state() == State.SUCCESS:
+ workload_key = self.active_workers.pop_by_id(job.job_id)
+ self.success(workload_key)
+ except Exception:
+ self.log.exception(
+ "Evicting Batch job %s after an unexpected error while
syncing it, "
+ "so that one broken job cannot abort the sync cycle and
stall task "
+ "submission for the whole executor.",
+ job.job_id,
+ )
+ self._evict_job(job.job_id)
+
+ def _evict_job(self, job_id: str) -> None:
+ """Remove a job from the collection and fail its workload, tolerating
corrupted bookkeeping."""
+ workload_key = self.active_workers.remove_job(job_id)
+ if workload_key is None:
+ return
+ try:
+ self.fail(workload_key)
Review Comment:
If we manage to find an active task, is there a chance it actually succeeded
in Batch? Should we auto fail everything here?
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##########
@@ -268,11 +268,30 @@ def sync_running_jobs(self):
self.log.debug("Active Workers: %s", describe_job_response)
for job in describe_job_response:
- if job.get_job_state() == State.FAILED:
- self._handle_failed_job(job)
- elif job.get_job_state() == State.SUCCESS:
- workload_key = self.active_workers.pop_by_id(job.job_id)
- self.success(workload_key)
+ try:
+ if job.get_job_state() == State.FAILED:
+ self._handle_failed_job(job)
+ elif job.get_job_state() == State.SUCCESS:
+ workload_key = self.active_workers.pop_by_id(job.job_id)
+ self.success(workload_key)
+ except Exception:
+ self.log.exception(
+ "Evicting Batch job %s after an unexpected error while
syncing it, "
+ "so that one broken job cannot abort the sync cycle and
stall task "
+ "submission for the whole executor.",
Review Comment:
I don't think this kind of justification is necessary in the log message.
--
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]
