This is an automated email from the ASF dual-hosted git repository.
uranusjr pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f102ad7f663 Lock TriggerCommsDecoder sync req-res cycle (#64882)
f102ad7f663 is described below
commit f102ad7f663d0e253f6f917c13326bf417c7c8c1
Author: Tzu-ping Chung <[email protected]>
AuthorDate: Wed Apr 15 15:43:56 2026 +0900
Lock TriggerCommsDecoder sync req-res cycle (#64882)
---
airflow-core/src/airflow/jobs/triggerer_job_runner.py | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index 7522349c461..f44037dc3e6 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -883,12 +883,14 @@ class TriggerCommsDecoder(CommsDecoder[ToTriggerRunner,
ToTriggerSupervisor]):
def _read_frame(self):
from asgiref.sync import async_to_sync
- return async_to_sync(self._aread_frame)()
+ with self._thread_lock:
+ return async_to_sync(self._aread_frame)()
def send(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
from asgiref.sync import async_to_sync
- return async_to_sync(self.asend)(msg)
+ with self._thread_lock:
+ return async_to_sync(self.asend)(msg)
async def _aread_frame(self):
try:
