ferruzzi commented on code in PR #61153:
URL: https://github.com/apache/airflow/pull/61153#discussion_r2738461118
##########
airflow-core/src/airflow/executors/local_executor.py:
##########
@@ -87,25 +94,31 @@ def _run_worker(
# Received poison pill, no more tasks to run
return
- if not isinstance(workload, workloads.ExecuteTask):
- raise ValueError(f"LocalExecutor does not know how to handle
{type(workload)}")
-
# Decrement this as soon as we pick up a message off the queue
with unread_messages:
unread_messages.value -= 1
- key = None
- if ti := getattr(workload, "ti", None):
- key = ti.key
- else:
- raise TypeError(f"Don't know how to get ti key from
{type(workload).__name__}")
- try:
- _execute_work(log, workload, team_conf)
+ # Handle different workload types
+ if isinstance(workload, workloads.ExecuteTask):
+ key = workload.ti.key
+ try:
+ _execute_work(log, workload, team_conf)
+ output.put((key, TaskInstanceState.SUCCESS, None))
+ except Exception as e:
+ log.exception("Task execution failed.")
+ output.put((key, TaskInstanceState.FAILED, e))
+
+ elif isinstance(workload, workloads.ExecuteCallback):
+ key = workload.callback.id
+ try:
+ _execute_callback(log, workload, team_conf)
+ output.put((key, TaskInstanceState.SUCCESS, None))
Review Comment:
I chewed this one a bit.... Its' either a SUCCESS or FAIL and I felt heavy
handed adding yet another state machine enum to repeat those values. I have an
idea I'll include in the next revision and see what you think.
--
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]
