dondaum commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r2923648811
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##########
@@ -447,6 +466,9 @@ def _submit_job_kwargs(
submit_job_api["containerOverrides"]["environment"].append(
{"name": "AIRFLOW_IS_EXECUTOR_CONTAINER", "value": "true"}
)
+ if queue:
+ submit_job_api["jobQueue"] = queue
Review Comment:
Cool. I've removed it for now. I'll then open another PR to add it.
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##########
@@ -127,26 +128,40 @@ def __init__(self, *args, **kwargs):
def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
from airflow.executors import workloads
- if not isinstance(workload, workloads.ExecuteTask):
+ if AIRFLOW_V_3_2_PLUS and isinstance(workload,
workloads.ExecuteCallback):
+ self.queued_callbacks[workload.callback.id] = workload
+ elif isinstance(workload, workloads.ExecuteTask):
+ ti = workload.ti
+ self.queued_tasks[ti.key] = workload
+ else:
raise RuntimeError(f"{type(self)} cannot handle workloads of type
{type(workload)}")
- ti = workload.ti
- self.queued_tasks[ti.key] = workload
def _process_workloads(self, workloads: Sequence[workloads.All]) -> None:
- from airflow.executors.workloads import ExecuteTask
+ from airflow.executors import workloads as wl
# Airflow V3 version
for w in workloads:
- if not isinstance(w, ExecuteTask):
+ if isinstance(w, wl.ExecuteTask):
+ tcommand = [w]
Review Comment:
changed.
--
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]
