[ChickenBenny]

• closes: execution_timeout on @task.stub tasks is not enforced #74716
• related: Language SDK runtimes do not handle execution_timeout in the task process #74717

---

Language SDK tasks never had their execution_timeout enforced, because only the Python task runner told the supervisor about it. Following the review, the timeout is now resolved once on the server and enforced by the supervisor for every task:

1. The supervisor takes execution_timeout only from TIRunContext. The Execution API resolves it from the serialized Dag and sends it for all tasks, Python included. This also covers mapped (.expand()) language SDK tasks, which previously got no timeout. A zero timeout means no timeout and a negative one has already elapsed, as in the Python task runner.

2. [core] default_task_execution_timeout now applies to language SDK tasks. A language SDK runtime cannot read the Airflow config, so the Dag processor fills the default into the serialized Dag's client_defaults when it stores the Dag, as it already does for the Dag-level defaults. A value the task sets wins over it.

3. SetExecutionTimeout is replaced with StartExecutionTimeout. The message no longer carries the seconds, and the Python task runner still sends it from the same place, right before execute(). SetExecutionTimeout shipped in 3.4.0b1 and 3.4.0b2 but is not in a final release yet, so the rename needs no supervisor schema migration if it lands before 3.4.0.

4. When the clock starts:

   • Python tasks: when the supervisor receives StartExecutionTimeout, so bundle loading and Dag parsing are not counted. Dag parsing already has [core] dagbag_import_timeout.
   • Language SDK tasks: right after the supervisor sends the runtime StartupDetails. Starting the runtime process is not counted; it has its own startup timeout.

   Either way, a task that has not stopped [core] killed_task_cleanup_time seconds after the timeout elapses is sent SIGTERM, then SIGKILL.

The execution_timeout field is added at 2026-10-30 in both the Execution API and the supervisor schema, so this needs backport-to-v3-4-test before 3.4.0.

Known gaps until the follow-ups land:

• The supervisor backstop needs an API server that includes this change. The 3.4.0 betas already serve API version 2026-10-30 without execution_timeout, so a worker with this change running against a beta API server gets no supervisor timeout. The Python task runner's own timer still applies.
• For a Python task, the runner's timer reads task.execution_timeout from the Dag file it parses, while the supervisor reads the serialized Dag. They can disagree when the bundle is not versioned and the Dag file changed after it was serialized, or when the Dag processor and the worker have different [core] default_task_execution_timeout, since each resolves that default when it imports the Dag file. If only the runner sees a timeout, the supervisor ignores StartExecutionTimeout; if only the serialized Dag has one, the runner never sends it. Follow-up 2 removes this gap.

Follow-ups (separate PRs):

1. Grace period semantics: language SDK tasks have no timer of their own, so nothing tells them about the timeout before they get SIGTERM at execution_timeout + killed_task_cleanup_time.
2. Make the Python task runner's timer read execution_timeout from TIRunContext instead of task.execution_timeout, so both timers use the serialized value. This is a small change that does not need to wait for 3. Worker-only settings of [core] default_task_execution_timeout then stop applying to Python tasks.
3. Decouple the Python task runner's SIGALRM timer, so that the runner only reacts to a signal and runs its own cleanup, as language SDK runtimes will.
4. Let a language SDK task opt out of [core] default_task_execution_timeout. The default only fills tasks that leave execution_timeout out, and the Go and TypeScript task options have no way to write an explicit null, so a task cannot ask for no timeout yet.
5. Fill the other task-level config defaults for language SDK tasks the same way.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[henry3260]

Thanks for the contribution! leave some comments

Doc nit: airflow-core/docs/core-concepts/tasks.rst:143 says the supervisor enforces the timeout "For tasks executed by the Python task runner". After this PR that is no longer true it also covers stub and native language-SDK tasks. Could we add a line saying so, or at least drop the "Python task runner" qualifier?

[jason810496]

Thanks for the PR.

Would it be better to align the execution_timeout at the core instead of fixing at each SDK?

By this way, we could treat the execution_timeout pin down at the Dag processing time as source of truth rather than retrieving the default value at the task runtime?

Additionally, for the SetExecutionTimeout introduced in #73806: would it be better to rename as StartExecutionTimeout and remove the execution_timeout field? -- cc @Andrushika

From my perspective, no matter it is Python task or any language SDK task, the execution_timeout should be figured out at the Execution API like current PR shape, and the runtime of each SDK should respect it from the TIRunContext. The coordinator level should be able to emit the signal once instead of implementing the signal for each SDK.

[ChickenBenny]

Thanks Jason for taking a look!

Quick update on where the PR is: killing a timed-out process is already handled in one place by the supervisor. The language SDK coordinators go through ActivitySubprocess too, so Go / TS / Java don't need any changes. The only thing that's still different is where the value comes from: Python still works it out itself and sends it to the supervisor.

Here's roughly what I'm planning to change:

1. Drop the is_stub-only check and send execution_timeout for all tasks
2. Have the Python task runner read the value from TIRunContext instead of working it out again
3. Rename SetExecutionTimeout to StartExecutionTimeout and drop timeout_seconds

A couple of things I'd like to check first:

• When the clock starts: after this change, the supervisor starts the clock as soon as the task starts and restarts it on the Start message. So if a Python task's bundle loading + Dag parsing takes too long, it might get killed before execute() even runs (right now that time isn't counted). Is that OK?
• Item 3 touches Andrushika's SetExecutionTimeout. Should that be a separate PR, or just keep it in this one?

[Andrushika]

Additionally, for the SetExecutionTimeout introduced in #73806: would it be better to rename as StartExecutionTimeout and remove the execution_timeout field? -- cc @Andrushika

Agree that it will be cleaner -- task runners in all SDKs don't need to work out the timeout by themselves anymore. But we still need to handle users' custom cleanup logic after the timeout.

To be more specific, make all lang SDKs signal-aware (like what #74690 is doing for Java), so that lang SDKs only need to listen to the signal and handle their own cleanup.

So if a Python task's bundle loading + Dag parsing takes too long, it might get killed before execute() even runs (right now that time isn't counted). Is that OK?

Hmmm I think a better solution could be making the supervisor only start the clock on the start message for Python tasks (the task runner sends it right before execute()), since the Dag file import already has its own timeout (see [core] dagbag_import_timeout, PR #74571, and the comment here: #73806 (comment)).
For lang SDK tasks, starting it when the runtime starts looks fine to me.

Item 3 touches Andrushika's SetExecutionTimeout. Should that be a separate PR, or just keep it in this one?

Thanks @ChickenBenny, I think it's fine to handle that in this PR. Since SetExecutionTimeout is already in 3.4.0b1, it would be nice to get the rename in before 3.4.0 together with the backport.

Also, two more things I want to mention:

1. The grace period semantics: Currently the supervisor side's timeout acts like a last resort, it lets the task runner have enough time to finish their cleanup first, and only sends the SIGTERM if the task runner itself cannot terminate properly. But lang SDK tasks have no timer of their own, so nothing tells them about the timeout during that grace period. They actually get SIGTERM at execution_timeout + killed_task_cleanup_time seconds.
2. If we finally decide to use the signal-cleanup shape I mentioned above: the Python side task runner still has its own timer (SIGALRM), so the timeout logic is not decoupled there yet. It might be a good fit to decouple it, same as the lang SDKs.

But these should be separate follow-up PRs. Just put it here as a reminder. I think we can first get the timeout working here and try to make it for 3.4.

[Andrushika]

FYI @ChickenBenny, this graph helped me understand the architecture between coordinators, executors, and the task runner a lot when I was a beginner. I hope this can help you as well :)

flowchart LR
    subgraph SP["Scheduler OS process"]
        subgraph LE["LocalExecutor object"]
            AQ["activity_queue<br/>IPC"]
            RQ["result_queue<br/>IPC"]
            WH["workers<br/>Process handles"]
        end
    end

    subgraph WP["Local worker OS process｜可重用"]
        LOOP["_run_worker()<br/>get loop"]
        CM["CoordinatorManager.for_queue()<br/>[sdk] queue_to_coordinator"]
        PYC["_PythonCoordinator<br/>queue 沒對應時的 fallback"]
        LSC["SubprocessCoordinator<br/>Executable／Java／Node"]
        subgraph SV["Supervisor｜supervise_task()"]
            SUP["ActivitySubprocess object<br/>WatchedSubprocess subtype"]
            PSUP["_PopenActivitySubprocess object<br/>ActivitySubprocess subtype"]
        end
    end

    subgraph TP["Task Runner child process｜每個 Python task 一個"]
        TR["task_runner.main()<br/>operator user code"]
    end

    subgraph LP["Lang runtime process｜每個 lang SDK task 一個"]
        GO["Go: bundle binary<br/>BundleRef.Serve()"]
        JV["Java: java -classpath ...<br/>Server.serve()"]
        TS["TS: node bundle.mjs<br/>bundle.serve()"]
    end

    AQ ==>|worker gets workload| LOOP
    LOOP -->|supervise_task per task| CM
    CM -->|queue not mapped| PYC
    CM -->|queue mapped| LSC
    PYC -.->|creates| SUP
    LSC -.->|creates| PSUP
    SUP ==>|puts outcome| RQ
    PSUP ==>|puts outcome| RQ
    WH -->|tracks／joins／terminates| LOOP
    SUP -->|forks／starts + waits| TR
    PSUP -->|launches runtime + waits| GO & JV & TS

    classDef focus fill:#fff3bf,stroke:#e67700,stroke-width:3px
    class SUP,PSUP focus

Loading

[jason810496]

the task runner sends it right before execute()
I think it's fine to handle that in this PR

Yes, both statements are correct.

What I imagined is 1) replace the Python side StartExecutionTimeout with StartExecutionTimeout while we still keep the same place to emit the message, where the bundle initialization already finished 2) for the Language SDK (_PopenActivitySubprocess), we should emit StartExecutionTimeout after the runtime starts, before StartupDetails

[ChickenBenny]

Thanks both! I've updated the PR:

1. The supervisor now takes execution_timeout only from TIRunContext (resolved from the serialized Dag), and the Execution API sends it for every task.
2. SetExecutionTimeout is replaced with StartExecutionTimeout, which no longer carries the seconds and is still sent from the same place.
3. Python tasks start the clock when the supervisor receives StartExecutionTimeout; language SDK tasks start it once the runtime has started, before StartupDetails.

Follow-ups, as discussed:

1. Grace period semantics for language SDK tasks.
2. Decoupling the Python SIGALRM timer. The runner's in-process timer still reads task.execution_timeout for now, so I'll move it to TIRunContext together with that, instead of in this PR.

[jason810496]

Thanks for the update, LGTM once address the following comments.

[jason810496]

Thanks for the update.

Following up on the single-writer thread: 0589fa1

Since StartExecutionTimeout has no payload, the value has to be stored at startup anyway, so how about one writer per field: _on_child_started sets _execution_timeout_seconds, and _arm_execution_timeout() takes no argument and only writes _execution_timeout_enforce_at.

[ChickenBenny]

Thanks! I've updated the PR and its description.

The description now has a "Known gaps" section. The main one: for a Python task, the runner's SIGALRM timer still reads task.execution_timeout from the Dag file, while the supervisor reads TIRunContext. If the two disagree (an unversioned bundle whose Dag file changed after it was serialized, or a different [core] default_task_execution_timeout on the Dag processor and the worker), the supervisor backstop is silently not armed.

To keep this PR small, I'll open a separate PR right after this one that makes the runner read TIRunContext too. It does not depend on decoupling the SIGALRM timer, so it won't have to wait for that.

[jason810496]

Nice, thanks for the update, LGTM now.

[Andrushika]

Thanks, LGTM!
Please keep working on the follow-up PRs if you would.

[jason810496]

Would you mind to rebase the PR when you have a moment? Thanks.

[ChickenBenny]

No problem, I've rebased it. The previous CI failure was a mypy error: after the rebase, ti_run assigned two different values to the same task variable. I've renamed mine to fix it. Fix with 287202e

[github-actions]

Backport successfully created: v3-4-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-4-test