fix frame mismatch in triggerer

[renat-sagut]

Fix for triggerers crashing due to frame mismatch. It happens when using Kafka assets for scheduling Dags.
Similar to CommsDecoder base class, synchronous send method is protected by the thread_lock, asynchronous send method is protected by the async_lock and the thread_lock.

• related: triggerer: TriggerCommsDecoder.asend() asyncio.Lock is not thread-safe — concurrent sync_to_async calls cause "Response read out of order" and fatal TriggerRunner crash #64620

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[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

[jscheffl]

With this change you are masking the implementation in CommdDecoder.asend() (in task-sdk/src/airflow/sdk/execution_time/comms.py:221) - but the problem of multi-threading and locking is a generic problem.

Have you considered applying the same change in CommdDecoder base class?
(I assume this is also used for the case of async tasks in worker with the new task run interface in Airflow 3.2, if we have async problems in triggerer while sending/receiving, will it not be the same for any other async task using the same in the base class?

[renat-sagut]

original solution is also overriding the same method

[jscheffl]

@ashb / @dabla Probably you are the experts for this. Might be OK but not sure if this has a reason not being consolidated with base class

[jscheffl]

re-opening, would be good to have an expert opinion. (I am not confident in this code area and fear I mis-understand the context of fix)

[jscheffl]

Cool! Thanks for the PR! Is there a chance to add a pytest in order to prevent a regression for the bug?

[potiuk]

@renat-sagut This PR has been converted to draft because it does not yet meet our Pull Request quality criteria.

Issues found:

• ❌ Unresolved review comments (3 threads (1 from a maintainer, 2 from other reviewers)): please walk through each unresolved review thread. Even if a suggestion looks incorrect or irrelevant — and some of them will be, especially any comments left by automated reviewers like GitHub Copilot — it is still the author's responsibility to respond: apply the fix, reply in-thread with a brief explanation of why the suggestion does not apply, or resolve the thread if the feedback is no longer relevant. Leaving threads unaddressed for weeks blocks the PR from moving forward.

What to do next:

• Walk through each unresolved review thread and respond as described above.
• Make sure static checks pass locally: prek run --from-ref main --stage pre-commit.
• Mark the PR as "Ready for review" when you're done.

Converting a PR to draft is not a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the Airflow Slack.

[potiuk]

Quick follow-up to the triage comment above — one clarification on the "Unresolved review comments" item:

Once you believe a thread has been addressed — whether by pushing a fix, or by replying in-thread with an explanation of why the suggestion doesn't apply — please mark the thread as resolved yourself by clicking the "Resolve conversation" button at the bottom of each thread. Reviewers don't auto-close their own threads, so an addressed-but-unresolved thread reads as "still waiting on the author" and keeps the PR from moving forward. The author doing the resolve-click is the expected convention on this project.