xBis7 commented on code in PR #69336:
URL: https://github.com/apache/airflow/pull/69336#discussion_r3960161323
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2631,6 +2631,15 @@ def supervise_task(
final_state=result.final_state,
)
return result.exit_code
+ except TaskAlreadyRunningError:
+ # Another worker is already running this task, so the server told
us to back off. This is
+ # normal -- it just means we were a duplicate that lost the race.
Our task never started any
+ # real work, so exit quietly instead of reporting a failure that
would look like a crash.
+ log.info(
+ "Task instance already running on another worker; standing
down without failing it",
+ workload_id=str(ti.id),
+ )
+ return 0
Review Comment:
As far as I know, Celery is the only executor currently handling
`TaskAlreadyRunningError` and when it does, it ignores the error.
This except is equally applied to all executors but it's also running before
Celery's except statement. So the error won't be handled in Celery anymore but
the behavior is conflicting.
Returning 0 means a success while Celery ignores the error.
See
https://github.com/dsuhinin/airflow/blob/dsuhinin/fix-scheduler-fail-running-ti-on-duplicate-dispatch/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py#L247-L253
and
https://github.com/dsuhinin/airflow/blob/dsuhinin/fix-scheduler-fail-running-ti-on-duplicate-dispatch/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py#L295-L301
I think the behavior should be consistent.
We can re-raise the error here and let it propagate to the executors. Celery
will catch it and treat the task failure as a no event.
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1581,7 +1581,17 @@ def process_executor_events(
)
)
- if ti_queued and not ti_requeued:
+ # A running task that's still sending heartbeats is alive -- a
worker is running it right now.
+ # This event is probably from a duplicate that already lost and
died, so don't fail the live
+ # run. If the task really did die, heartbeat detection will fail
it once the heartbeat stops.
+ heartbeat_timeout = conf.getint("scheduler",
"task_instance_heartbeat_timeout")
+ ti_alive = (
+ ti.state == TaskInstanceState.RUNNING
+ and ti.last_heartbeat_at is not None
+ and ti.last_heartbeat_at >= timezone.utcnow() -
timedelta(seconds=heartbeat_timeout)
+ )
+
+ if ti_queued and not ti_requeued and not ti_alive:
Review Comment:
I think a metric to monitor how many times the `ti_alive` condition has been
satisfied, might be useful for the future. This could be a follow-up.
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1581,7 +1581,17 @@ def process_executor_events(
)
)
- if ti_queued and not ti_requeued:
+ # A running task that's still sending heartbeats is alive -- a
worker is running it right now.
+ # This event is probably from a duplicate that already lost and
died, so don't fail the live
+ # run. If the task really did die, heartbeat detection will fail
it once the heartbeat stops.
+ heartbeat_timeout = conf.getint("scheduler",
"task_instance_heartbeat_timeout")
Review Comment:
We are in a `for ti in tis:` loop and we are reading the config again for
every iteration. We should read it only once, outside of the loop.
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1516,7 +1516,17 @@ def process_executor_events(
)
)
- if ti_queued and not ti_requeued:
+ # A running task that's still sending heartbeats is alive -- a
worker is running it right now.
+ # This event is probably from a duplicate that already lost and
died, so don't fail the live
+ # run. If the task really did die, heartbeat detection will fail
it once the heartbeat stops.
+ heartbeat_timeout = conf.getint("scheduler",
"task_instance_heartbeat_timeout")
+ ti_alive = (
+ ti.state == TaskInstanceState.RUNNING
Review Comment:
@ashb I went over the code and it looks to me that @dsuhinin is right. These
are different.
The state in your code snippet is the state that we get from the executor as
a response.
```python
state, info = event_buffer.pop(buffer_key)
if state in (TaskInstanceState.QUEUED, TaskInstanceState.RUNNING):
ti.external_executor_id = info
cls.logger().info("Setting external_executor_id for %s to %s", ti, info)
continue
```
https://github.com/dsuhinin/airflow/blob/bb496e9a9be955a03566e5dc5eaf7b681b9a4fc6/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1513
While the state in these changes, is the one that we read from the DB.
```python
locked_query = with_row_locks(query, of=TI, session=session,
skip_locked=True)
tis: Iterator[TI] = session.scalars(locked_query)
for ti in tis:
```
https://github.com/dsuhinin/airflow/blob/bb496e9a9be955a03566e5dc5eaf7b681b9a4fc6/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1497-L1499
--
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]
