Fix trigger kwargs decoding for asset watcher triggers

[shaealh]

Fixes #65973

Trigger kwargs for asset watcher triggers can pass through both BaseSerialization and SDK serde. When this happens, old serialization marker keys can be stored as the stringified enum names Encoding.TYPE and Encoding.VAR, which prevents triggerer-side cleanup from decoding nested kwargs before trigger execution.

This updates trigger kwargs decoding to normalize those stringified marker keys and recursively decode nested serialized kwargs before trigger instantiation.

Tests:

• AIRFLOW_HOME=/tmp/airflow-65973-test uv run pytest airflow-core/tests/unit/jobs/test_triggerer_job.py::TestTriggerRunner::test_trigger_kwargs_cleanup_decodes_stringified_encoding_keys airflow-core/tests/unit/jobs/test_triggerer_job.py::TestTriggerRunner::test_trigger_kwargs_serialization_cleanup
• uv run ruff check airflow-core/src/airflow/serialization/decoders.py airflow-core/tests/unit/jobs/test_triggerer_job.py

[dshk0718]

Hello @ashb @bolkedebruin,
Will this be included in the upcoming Airflow 3.2.2 release? I am hoping that I can upgrade to version 3.2.x from version 3.1.x soon, and without this fix, I can't.

[ashb]

Hello @ashb @bolkedebruin,

Will this be included in the upcoming Airflow 3.2.2 release? I am hoping that I can upgrade to version 3.2.x from version 3.1.x soon, and without this fix, I can't.

Given this is not yet merged there is no way it can new included in any release, especially not one released already

[ashb]

@jedcunningham @amoghrajesh can you take a look please?

[dshk0718]

Hello @ashb @bolkedebruin,
Will this be included in the upcoming Airflow 3.2.2 release? I am hoping that I can upgrade to version 3.2.x from version 3.1.x soon, and without this fix, I can't.

Given this is not yet merged there is no way it can new included in any release, especially not one released already

Thanks for the quick reply. What do you mean by "released already"? Are you talking about version 3.2.1 here? I do not see version 3.2.2 in the release notes yet on the Airflow documentation site. Also, if I am not mistaken, the milestone for version 3.2.2 is not complete yet. That's why I asked if this would be available in the 3.2.2 release.

[ashb]

My bad, getting version numbers mixed up! Same though, if it's reviewed and merged, it'll be in the next bug fix release

[dshk0718]

Hello,

Sorry to bother y'all here again, but can we have the merge conflicts resolved, @shaealh? And then, can we have this reviewed? I am really hoping that this will be included in the next 3.2.x release, so we can finally upgrade to Airflow 3.2.x version without worrying about breaking any of our production functionalities and finally get to enjoy all the great improvements that have been added to v3.2.0+ versions.

[potiuk]

@shaealh — Removing the ready for maintainer review label and converting back to draft. The branch now has merge conflicts with main that surfaced after the label was added.

The label's contract is that the PR is ready for maintainer review — a regression like this means the PR temporarily isn't. Rebase your branch onto the latest main, resolve conflicts, then mark "Ready for review" again to re-enter the queue.

git fetch upstream main && git rebase upstream/main, resolve, git push --force-with-lease. See the working-with-git docs.

No rush.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[shaealh]

@ashb @jedcunningham @amoghrajesh anything I can do on my end to help get your review so I can prepare to merge?

[dshk0718]

@ashb @bolkedebruin @jedcunningham @potiuk @amoghrajesh
Is it possible to have the issue I've created (#65973) and this pull request included in either the Airflow 3.2.3 milestone or the Airflow 3.3.0 milestone, and then get this reviewed and approved?
Given that Airflow 3.2.2 has been released without this fix merged in, I still probably won't be able to upgrade because the DAGs I've written heavily rely on the KafkaMessageQueueTrigger and the keyword arguments that I am providing to the custom apply function that I am using.

[shaealh]

@ashb @bolkedebruin @jedcunningham @potiuk @amoghrajesh
Can you take a look when you get a chance

[shaealh]

@vatsrahul1001 @ashb @bolkedebruin @jedcunningham @potiuk @amoghrajesh
Any feedback based on your review? Thanks

[shaealh]

@potiuk A couple of months under review, any updates? Thanks

[shaealh]

@bolkedebruin @ashb can you approve for merge?

[Vamsi-klu]

Dug into smart_decode_trigger_kwargs and the root cause checks out. The old Encoding.TYPE not in d guard compares against __type, but the asset watcher kwargs that went through both BaseSerialization and serde persisted their keys as the stringified enum names Encoding.TYPE/Encoding.VAR, so the guard fell through and the raw marker dict got handed straight to the trigger constructor. _normalize_stringified_encoding_keys remapping those keys back to the enum members before the Encoding.VAR in d check, plus the recursion into non-wrapper dicts, lines up with the shape from #65973, and a normally serialized __type/__var payload still lands on the same deserialize path so nothing regresses. One nit, the recursive branch re-normalizes children the top-level call already walked, so the tree gets traversed twice, but it's idempotent and the kwargs are tiny so not worth blocking. Nice that the test drives it through runner.create_triggers() rather than poking the decoder directly.

[shaealh]

Thanks for the detailed review @Vamsi-klu and for validating the root cause and fix. Good catch on the redundant traversal. Since it is idempotent and the kwargs are small, I’ll leave it as is unless a maintainer (@potiuk, @ashb, @bolkedebruin) prefers that it be cleaned up before merging.