Handle task timeouts (`execution_timeout`) at supervisor

[amoghrajesh]

Body

Right now we handle these at the task runner (fork: https://github.com/apache/airflow/blob/main/task-sdk/src/airflow/sdk/execution_time/task_runner.py#L1149-L1165) instead of handling it at the supervisor. We should move the check to the supervisor level.

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.

[ashb]

Since the timeout is defined in the dag we'll need to communicate the timeout to the supervisor- possibly by the Run request (but there's no point sending it to the API server that I can immediately think of)

[potiuk]

Another option is to make a second fork:

supervisor -> timeout fork -> task fork

And have the timeout fork only handle the timeout via ALRM signal. And the timeout fork could do dag parsing and deciding what to run before forking the task fork. This way it can retrieve timeout from DAG definition, and then (because it will fork) - the resulting task fork will already have the task parsed in memory.

While forks are not as "good" for Python interpreter for memory as for regular c-code (because of reference counting almost all memory is very quickly copied-on-write. this is not the case when the fork will merely join the forked process and react to ALRM signal.

Would be worth seeing if this approach might be good for memory usage - but it has the "isolation" property that is rather nice. We can have single ALRM handler in the "timeout" fork rather than handling more complex cases of supervisor waiting for the timeout additionally. But also it might be an overkill - I thought I will just mention it as we discussed that option as potential implementation in 2.0