Enforce execution_timeout for tasks run by a language SDK

[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 from the serialized Dag 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.
2. 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 but is not in a final release yet, so the rename needs no supervisor schema migration if it lands before 3.4.0.
3. 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: once the runtime has started, right before StartupDetails is sent.

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.

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. Decouple the Python task runner's SIGALRM timer. It still reads task.execution_timeout instead of TIRunContext; that will move together with the decoupling.

[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?

[henry3260]

get_language_sdk_execution_timeout uses a falsy check, so a negative
execution_timeout passes through: timedelta(seconds=-1) is truthy, returns
-1.0, and _arm_execution_timeout lands the deadline at
monotonic() - 1 + KILLED_TASK_CLEANUP_TIME SIGTERM 59s later rather than at
once. bases/operator.py only type-checks the field, so this is reachable.

The Python task runner fast-fails instead (if timeout_seconds <= 0: raise AirflowTaskTimeout(), right before it sends SetExecutionTimeout), which is why
the supervisor never sees a non-positive value today.

• Treat non-positive as no timeout, which keeps the fix inside the helper and
  matches what the docstring already says about zero:

  if (task := get_stub_task(dag_bag, ti, session=session)) is None:
      return None
  timeout_seconds = task.execution_timeout.total_seconds() if task.execution_timeout else 0
  if timeout_seconds <= 0:
      return None
  return timeout_seconds

[ChickenBenny]

I fixed it in 658b814, but went a slightly different way from your suggestion: a negative value is clamped to 0.0 instead of returning None. This aligns with how the Python task runner handles it, where a negative timeout fails the task right away rather than disabling the timeout:

airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py

Lines 2318 to 2326 in f3784d2

	if task.execution_timeout:
	from airflow.sdk.execution_time.timeout import timeout
	# TODO: handle timeout in case of deferral
	timeout_seconds = task.execution_timeout.total_seconds()
	try:
	# It's possible we're already timed out, so fast-fail if true
	if timeout_seconds <= 0:
	raise AirflowTaskTimeout()

[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]

nit: Let's retrieve the SerializedOperator via get_task and pass it down to arg_bindings and execution_timeout to avoid the duplicated prob overhead.

[ChickenBenny]

Fix with 5c03237

[jason810496]

IIUC, we need to fill the task level execution_timeout at DagSerialization.fill_config_defaults.
We can keep execution_timeout for now and follow-up for rest of the task-level default arguments.

[ChickenBenny]

Fix with 2bb4a67

[jason810496]

IIUC, we don't need to set the _execution_timeout_seconds here. Let's keep the _arm_execution_timeout as single writer to mutate the attribute state.

[ChickenBenny]

Fix with 0589fa1