Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi commented on PR #62984: URL: https://github.com/apache/airflow/pull/62984#issuecomment-4443515263 That was a long slog, and I appreciate the work you've been doing on this and the surrounding efforts. Cheers. -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi merged PR #62984: URL: https://github.com/apache/airflow/pull/62984 -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi commented on code in PR #62984: URL: https://github.com/apache/airflow/pull/62984#discussion_r3230521782 ## 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: Review Comment: Yeah, I think this is resolved for now. The narrower QueueableWorkload type from #63491 will replace this once that PR merges and the min version bumps. For now, `workloads.All` is correct and consistent with the other executors. -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on PR #62984: URL: https://github.com/apache/airflow/pull/62984#issuecomment-4428899515 > @dondaum - Resolve that merge conflict and I'll approve once the CI is green 👍 done. -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi commented on PR #62984: URL: https://github.com/apache/airflow/pull/62984#issuecomment-4426292179 @dondaum - Resolve that merge conflict and I'll approve once the CI is green :+1: -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984: URL: https://github.com/apache/airflow/pull/62984#discussion_r3152573845 ## 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: Review Comment: Thanks for another look at this. In my opinion we can solve it. The relevant comment can be found at https://github.com/apache/airflow/pull/63657/changes#r3113924410. -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
o-nikolas commented on code in PR #62984: URL: https://github.com/apache/airflow/pull/62984#discussion_r3149348122 ## 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: Review Comment: What is the state of this thread? Is it still open or can we resolve this one? -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3139328444
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
Review Comment:
Oh you already did both. Sorry. I'll close mine. Cheers.
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3139319582
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
Review Comment:
https://github.com/apache/airflow/pull/65814
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3139319582
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
Review Comment:
Fix it in Base and Celery: https://github.com/apache/airflow/pull/65814
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3136588736
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
Review Comment:
Okay. 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3133921512
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
Review Comment:
Sorry for the delay. I've asked them all to make the same change, you
don't need to handle it all. I believe Celery is the only one that isn't being
fixed at the moment and we can do that one when we do Base.
I don't think we need to worry about the parameter name mismatch as much as
you are. It's a "private" method with only one parameter so it is only ever
being called by positional args and the name doesn't matter.
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3130329210
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
Review Comment:
@ferruzzi Any thoughts on this ?
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3119818589
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
Review Comment:
They all use
```Python
def _process_workloads(self, workloads: Sequence[workloads.All]) -> None:
...
```
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3119809027
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
Review Comment:
Of course I can. But this change has a certain explosion radius. I have to
change too
- AWSLambdaExecutor
- AWSEcsExecutor
- EdgeExecutor
- KubernetesExecutor
- CeleryExecutor
Maybe this would stretch the scope of the change? WDYT?
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3119521170
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
Review Comment:
Both. If you name the parameter `workload_items` then you get rid of the
name shadowing and don't have to do an `import ... as`. This wasn't
something I considered when I wrote BaseExecutor's method and everyone is
running into it. In fact, could you change that parameter name in BaseExecutor
as well, so everyone else doesn't have this same hiccup?
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3119497654
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -337,18 +357,18 @@ def attempt_submit_jobs(self):
self.log.error(
(
"This job has been unsuccessfully attempted too
many times (%s). "
-"Dropping the task. Reason: %s"
+"Dropping the workload. Reason: %s"
),
attempt_number,
failure_reason,
)
self.log_task_event(
event="batch job submit failure",
extra=f"This job has been unsuccessfully attempted too
many times ({attempt_number}). "
-f"Dropping the task. Reason: {failure_reason}",
-ti_key=key,
+f"Dropping the workload. Reason: {failure_reason}",
+ti_key=workload_key,
)
-self.fail(key=key)
+self.fail(key=workload_key)
Review Comment:
I think this looks right. Resolving.
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984: URL: https://github.com/apache/airflow/pull/62984#discussion_r3116309577 ## providers/amazon/src/airflow/providers/amazon/aws/executors/batch/utils.py: ## @@ -43,7 +52,7 @@ class BatchQueuedJob: """Represents a Batch job that is queued. The job will be run in the next heartbeat.""" -key: TaskInstanceKey +key: BatchJobWorkloadKey command: CommandType queue: str Review Comment: Good catch. Changed. ## providers/amazon/tests/unit/amazon/aws/executors/batch/test_batch_executor.py: ## @@ -269,6 +270,98 @@ def test_task_sdk(self, running_state_mock, mock_airflow_key, mock_executor, moc assert job_id == ARN1 running_state_mock.assert_called_once_with(workload.ti.key, ARN1) [email protected](not AIRFLOW_V_3_3_PLUS, reason="Test requires Airflow 3.2+") 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3116307804
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
Review Comment:
Anyway I changed to `Sequence[workloads.All]`
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
# Airflow V3 version
Review Comment:
done.
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984: URL: https://github.com/apache/airflow/pull/62984#discussion_r3116305877 ## providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py: ## @@ -88,14 +97,15 @@ class AwsBatchExecutor(BaseExecutor): """ supports_multi_team: bool = True +supports_callbacks: bool = True Review Comment: Thanks. Changed it. -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3116303914
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -337,18 +357,18 @@ def attempt_submit_jobs(self):
self.log.error(
(
"This job has been unsuccessfully attempted too
many times (%s). "
-"Dropping the task. Reason: %s"
+"Dropping the workload. Reason: %s"
),
attempt_number,
failure_reason,
)
self.log_task_event(
event="batch job submit failure",
extra=f"This job has been unsuccessfully attempted too
many times ({attempt_number}). "
-f"Dropping the task. Reason: {failure_reason}",
-ti_key=key,
+f"Dropping the workload. Reason: {failure_reason}",
+ti_key=workload_key,
)
-self.fail(key=key)
+self.fail(key=workload_key)
Review Comment:
I've already rebased it, but in my opinion no further changes are needed. Or
am I wrong?
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
Review Comment:
Thanks. Changed it.
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3116156421
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
Review Comment:
I suspect you just want to change the type from `WorkflowTypes` to
`workload.All`. Right? Because `BaseExecutor `does not have a method parameter
named `workload_items`
https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/executors/base_executor.py#L293
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3114172659
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
Review Comment:
My mistake, sorry. We can't do that until min_ver hits 3.2. Let's go with
one standard replacement across all of these though. Try this:
```python
def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
from airflow.executors import workloads
if isinstance(workload, workloads.ExecuteTask):
self.queued_tasks[workload.ti.key] = workload
return
if AIRFLOW_V_3_3_PLUS and isinstance(workload,
workloads.ExecuteCallback):
self.queued_callbacks[workload.callback.key] = workload
return
raise RuntimeError(f"{type(self)} cannot handle workloads of type
{type(workload)}")
```
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3114005158
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -88,14 +97,15 @@ class AwsBatchExecutor(BaseExecutor):
"""
supports_multi_team: bool = True
+supports_callbacks: bool = True
Review Comment:
This should be gated
```suggestion
if AIRFLOW_V_3_3_PLUS:
supports_callbacks: bool = True
```
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
Review Comment:
Double check, but you should be able to drop this entirely and rely on
BaseExecutor's version
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/utils.py:
##
@@ -43,7 +52,7 @@
class BatchQueuedJob:
"""Represents a Batch job that is queued. The job will be run in the next
heartbeat."""
-key: TaskInstanceKey
+key: BatchJobWorkloadKey
command: CommandType
queue: str
Review Comment:
Double check me here, but `_process_workloads` can set queue = None so this
should be
```suggestion
queue: str | None
```
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
Review Comment:
Same comment as https://github.com/apache/airflow/pull/63657/:
```suggestion
def _process_workloads(self, workload_items: Sequence[workloads.All])
-> None:
from airflow.executors import workloads
```
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -124,29 +134,48 @@ def __init__(self, *args, **kwargs):
fallback=CONFIG_DEFAULTS[AllBatchConfigKeys.MAX_SUBMIT_JOB_ATTEMPTS],
)
-def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+def queue_workload(self, workload: WorkloadTypes, session: Session | None)
-> None:
from airflow.executors import workloads
-if not isinstance(workload, workloads.ExecuteTask):
+if AIRFLOW_V_3_3_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
+def _process_workloads(self, workloads: Sequence[WorkloadTypes]) -> None:
+from airflow.executors import workloads as wl
# Airflow V3 version
Review Comment:
We can drop this comment while you are in there please.
##
providers/amazon/tests/unit/amazon/aws/executors/batch/test_batch_executor.py:
##
@@ -269,6 +270,98 @@ def test_task_sdk(self, running_state_mock,
mock_airflow_key, mock_executor, moc
assert job_id == ARN1
running_state_mock.assert_called_once_with(workload.ti.key, ARN1)
[email protected](not AIRFLOW_V_3_3_PLUS, reason="Test requires Airflow
3.2+")
Review Comment:
Here and below, update the version in the reas
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984: URL: https://github.com/apache/airflow/pull/62984#discussion_r3111023143 ## 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: Review Comment: Since https://github.com/apache/airflow/pull/62645 is now merged, I can narrow down the types. -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r3111001483
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -337,18 +357,18 @@ def attempt_submit_jobs(self):
self.log.error(
(
"This job has been unsuccessfully attempted too
many times (%s). "
-"Dropping the task. Reason: %s"
+"Dropping the workload. Reason: %s"
),
attempt_number,
failure_reason,
)
self.log_task_event(
event="batch job submit failure",
extra=f"This job has been unsuccessfully attempted too
many times ({attempt_number}). "
-f"Dropping the task. Reason: {failure_reason}",
-ti_key=key,
+f"Dropping the workload. Reason: {failure_reason}",
+ti_key=workload_key,
)
-self.fail(key=key)
+self.fail(key=workload_key)
Review Comment:
fixed here https://github.com/apache/airflow/pull/65392
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r2945619055
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -337,18 +357,18 @@ def attempt_submit_jobs(self):
self.log.error(
(
"This job has been unsuccessfully attempted too
many times (%s). "
-"Dropping the task. Reason: %s"
+"Dropping the workload. Reason: %s"
),
attempt_number,
failure_reason,
)
self.log_task_event(
event="batch job submit failure",
extra=f"This job has been unsuccessfully attempted too
many times ({attempt_number}). "
-f"Dropping the task. Reason: {failure_reason}",
-ti_key=key,
+f"Dropping the workload. Reason: {failure_reason}",
+ti_key=workload_key,
)
-self.fail(key=key)
+self.fail(key=workload_key)
Review Comment:
Great find. I followed the relevant code. The scheduler uses this log queue
to write a `Log()` entry. The log itself can be initialized without a task
instance, but the Executor method expects a task instance key `def
log_task_event(self, *, event: str, extra: str, ti_key: TaskInstanceKey)`.
So I'm wondering whether we should either adjust the ` def log_task_event`()
to accept both keys, which would also perhaps require a change to a different
executor, or whether we should remove the callback from this log queue.
@ferruzzi any thoughts on it ?
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
shivaam commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r2937493282
##
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##
@@ -337,18 +357,18 @@ def attempt_submit_jobs(self):
self.log.error(
(
"This job has been unsuccessfully attempted too
many times (%s). "
-"Dropping the task. Reason: %s"
+"Dropping the workload. Reason: %s"
),
attempt_number,
failure_reason,
)
self.log_task_event(
event="batch job submit failure",
extra=f"This job has been unsuccessfully attempted too
many times ({attempt_number}). "
-f"Dropping the task. Reason: {failure_reason}",
-ti_key=key,
+f"Dropping the workload. Reason: {failure_reason}",
+ti_key=workload_key,
)
-self.fail(key=key)
+self.fail(key=workload_key)
Review Comment:
When a callback workload exceeds max submit attempts, log_task_event is
called with ti_key=workload_key. For callbacks, this key is a string UUID, not
a TaskInstanceKey named tuple, which will cause errors since
Log(task_instance=...) expects a TaskInstanceKey.
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984: URL: https://github.com/apache/airflow/pull/62984#discussion_r2930205688 ## 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: Review Comment: Let's say we make this conditional input work here. We are still limited by the `BaseExecutor` [type ](https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/executors/base_executor.py#L222) having this ```Python def queue_workload(self, workload: workloads.All, session: Session) -> None: ... ``` -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984: URL: https://github.com/apache/airflow/pull/62984#discussion_r2930205688 ## 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: Review Comment: Let's say we make this conditional input work here. We are still limited by the `BaseExecutor` type having this ```Python def queue_workload(self, workload: workloads.All, session: Session) -> None: ... ``` -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
ferruzzi commented on code in PR #62984: URL: https://github.com/apache/airflow/pull/62984#discussion_r2928103742 ## 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: Review Comment: @onikolas is likely thinking of `SchedulerWorkload` from workloads/types.py, but @dondaum may be right, that isn't going to be in 2.11 so we can't use that until we bunmp the min_ver. I think we could do a conditional import though: ```python workload_type_hint = workloads.All if airflow version > 3.2: from airflow.executors.workloads.types import SchedulerWorkload workload_type_hint = SchedulerWorkload ``` And that will force it to get cleaned up later when we pin the versions up?? -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
o-nikolas commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r2927913043
##
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:
Much appreciated :slightly_smiling_face:
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
o-nikolas commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r2927911493
##
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:
Review Comment:
Even for just typing? @ferruzzi thoughts?
##
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:
Thanks!
--
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984: URL: https://github.com/apache/airflow/pull/62984#discussion_r2923407039 ## 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: Review Comment: Thank you for the review. Are you referring to types `ExecuteTask `and `ExecuteCallback`? https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/executors/workloads/__init__.py I think we can only use it if we pin the provider's Airflow core dependency to when this change was implemented. This happened here https://github.com/apache/airflow/pull/61153. If we want to maintain backwards compatibility, I can't use the new types.. Currently the Amazon provider is `apache-airflow>=2.11.0` -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984: URL: https://github.com/apache/airflow/pull/62984#discussion_r2923407039 ## 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: Review Comment: Thank you for the review. Are you referring to types `ExecuteTask `and `ExecuteCallback`? https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/executors/workloads/__init__.py I think we can only use it if we pin the provider's Airflow core dependency to when this change was implemented. This happened here https://github.com/apache/airflow/pull/61153. If we want to maintain backwards compatibility (currently the Amazon provider is `apache-airflow>=2.11.0` -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on PR #62984: URL: https://github.com/apache/airflow/pull/62984#issuecomment-4045639800 > I still see a lot of other methods with names, types, variables, comments all mentioning tasks where it should be workloads that need cleaning up. > > @ferruzzi should have a look too of course I changed the reference from task to workload where I could make a change. There is a method `try_adopt_task_instances() `which I have not changed since it is part of the base class. But technically it should work on both tasks and callbacks. -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
dondaum commented on code in PR #62984: URL: https://github.com/apache/airflow/pull/62984#discussion_r2923407039 ## 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: Review Comment: Thanks for the review. Are you refering to `ExecuteTask` and `ExecuteCallback` ? https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/executors/workloads/__init__.py I think we can only use it, if we pin the provider airflow-core depdency to whenever this change was implemented. This happend here https://github.com/apache/airflow/pull/61153. If we want to keep backwards compatiability (currently the amazon provider is pins `apache-airflow>=2.11.0"` Thank you for the review. Are you referring to types `ExecuteTask `and `ExecuteCallback`? https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/executors/workloads/__init__.py I think we can only use it if we pin the provider's Airflow core dependency to when this change was implemented. This happened here https://github.com/apache/airflow/pull/61153. If we want to maintain backwards compatibility (currently the Amazon provider is `apache-airflow>=2.11.0` -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
potiuk commented on PR #62984: URL: https://github.com/apache/airflow/pull/62984#issuecomment-4043132065 @dondaum This PR has a few issues that need to be addressed before it can be reviewed — please see our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria). **Issues found:** - :warning: **Unresolved review comments**: This PR has 3 unresolved review threads from maintainers. Please review and resolve all inline review comments before requesting another review. You can resolve a conversation by clicking 'Resolve conversation' on each thread after addressing the feedback. See [pull request guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst). > **Note:** Your branch is **278 commits behind `main`**. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results. **What to do next:** - The comment informs you what you need to do. - Fix each issue, then mark the PR as "Ready for review" in the GitHub UI - but only after making sure that all the issues are fixed. - Maintainers will then proceed with a normal review. Please address the issues above and push again. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack). -- 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]
Re: [PR] feat: add callback support to aws batch executor [airflow]
o-nikolas commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r2921507643
##
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:
Please use full names, the clarity is worth 4 more characters :)
##
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:
Let's not conflate this change along with supporting callbacks. So far this
executor has not been opinionated about the `queue` parameter. I like your
suggestion, but let's discuss it in it's own PR.
##
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:
Review Comment:
If you don't mind can you narrow down the type for workload here? I think
there is a new type that @ferruzzi made basically anywhere you need `task |
callback` as a type.
--
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]
