You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "kaxil (via GitHub)" <gi...@apache.org> on 2026/04/01 17:22:32 UTC

[PR] Fix deferred task resume failure when worker is older than server [airflow]

kaxil opened a new pull request, #64598:
URL: https://github.com/apache/airflow/pull/64598

   ## Summary
   
   - Workers on task-sdk < 1.2 crash with `KeyError: __var` when resuming deferred tasks whose trigger has fired on a 3.2 server
   - Add a Cadwyn response converter on `TIRunContext` that wraps `next_kwargs` in BaseSerialization format for old API versions
   
   ## Context
   
   After #59711 switched trigger kwargs to SDK serde, `handle_event_submit` re-serializes `next_kwargs` as plain dicts. Old workers only know `BaseSerialization.deserialize()`, which requires `{"__type": "dict", "__var": {...}}` wrapping on all dicts -- so they crash with `KeyError: __var`.
   
   This was [flagged during review of #59711](https://github.com/apache/airflow/pull/59711#discussion_r2650977380) -- the compat shim was added for the deserialize path but not the serialize path.
   
   ## Approach
   
   Instead of changing `trigger.py` to use `BaseSerialization.serialize()` (which regresses the DB to old format), the fix adds a response converter in the Execution API versioning layer (`ModifyDeferredTaskKwargsToJsonValue`). This:
   
   - Keeps the DB in SDK serde format (the direction we're moving)
   - Translates at the API boundary for old workers -- consistent with how all other backward-compat changes work via Cadwyn
   - Short-circuits when data already has `__type`/`__var` keys (rolling upgrade with old data in DB)
   - Gets removed automatically when old API versions are dropped


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix deferred task resume failure when worker is older than server [airflow]

Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 merged PR #64598:
URL: https://github.com/apache/airflow/pull/64598


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix deferred task resume failure when worker is older than server [airflow]

Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #64598:
URL: https://github.com/apache/airflow/pull/64598#discussion_r3071458589


##########
airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_04_06.py:
##########
@@ -118,6 +118,34 @@ class ModifyDeferredTaskKwargsToJsonValue(VersionChange):
         schema(TIDeferredStatePayload).field("next_kwargs").had(type=dict[str, Any]),
     )
 
+    @convert_response_to_previous_version_for(TIRunContext)  # type: ignore[arg-type]
+    def convert_next_kwargs_to_base_serialization(response: ResponseInfo) -> None:  # type: ignore[misc]
+        """
+        Convert next_kwargs from SDK serde format to BaseSerialization format for old workers.
+
+        Old workers (task-sdk < 1.2) only know BaseSerialization.deserialize(), which requires
+        dicts wrapped as {"__type": "dict", "__var": {...}}. SDK serde produces plain dicts that
+        BaseSerialization cannot parse, causing KeyError on __var.
+
+        We must deserialize SDK serde first to recover native Python objects (datetime,
+        timedelta, etc.), then re-serialize with BaseSerialization so old workers get
+        proper typed values instead of raw {"__classname__": ...} dicts.
+        """
+        next_kwargs = response.body.get("next_kwargs")
+        if next_kwargs is None:
+            return
+
+        from airflow.sdk.serde import deserialize
+        from airflow.serialization.serialized_objects import BaseSerialization
+
+        try:
+            plain = deserialize(next_kwargs)
+        except (ImportError, KeyError, AttributeError, TypeError):
+            # Already in BaseSerialization format (rolling upgrade, old data in DB)
+            return
+
+        response.body["next_kwargs"] = BaseSerialization.serialize(plain)

Review Comment:
   Thank you, this makes sense and was a miss from me!



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix deferred task resume failure when worker is older than server [airflow]

Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on PR #64598:
URL: https://github.com/apache/airflow/pull/64598#issuecomment-4175475457

   E2E failure looks unrelated. Failing for others' PR as well.


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix deferred task resume failure when worker is older than server [airflow]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #64598:
URL: https://github.com/apache/airflow/pull/64598#issuecomment-4175484671

   ### Backport successfully created: v3-2-test
   
   Note: As of [Merging PRs targeted for Airflow 3.X](https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#merging-prs-targeted-for-airflow-3x)
   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](https://apache-airflow.slack.com/archives/C03G9H97MM2) Slack channel.
   
   <table>
           <tr>
           <th>Status</th>
           <th>Branch</th>
           <th>Result</th>
       </tr>
       <tr>
           <td>✅</td>
           <td>v3-2-test</td>
           <td><a href="https://github.com/apache/airflow/pull/64619"><img src="https://img.shields.io/badge/PR-64619-blue" alt="PR Link"></a></td>
       </tr>
   </table>


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix deferred task resume failure when worker is older than server [airflow]

Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on PR #64598:
URL: https://github.com/apache/airflow/pull/64598#issuecomment-4174884063

   e2e test failing  `tests/airflow_e2e_tests/basic_tests/test_basic_dag_operations.py::TestBasicDagFunctionality::test_xcom_value FAILED [  2%]`


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org