This is an automated email from the ASF dual-hosted git repository.
jscheffl 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 466eb1c0ffb Fix deadlock and add unit test (#66144)
466eb1c0ffb is described below
commit 466eb1c0ffb8374aeabb4d733db74a675ea11406
Author: AutomationDev85 <[email protected]>
AuthorDate: Thu Apr 30 22:07:31 2026 +0200
Fix deadlock and add unit test (#66144)
Co-authored-by: AutomationDev85 <AutomationDev85>
---
.../src/airflow/providers/edge3/cli/worker.py | 22 ++++---
.../edge3/tests/unit/edge3/cli/test_worker.py | 70 +++++++++++++++++++++-
2 files changed, 83 insertions(+), 9 deletions(-)
diff --git a/providers/edge3/src/airflow/providers/edge3/cli/worker.py
b/providers/edge3/src/airflow/providers/edge3/cli/worker.py
index 799b34aefd4..c47c2063aad 100644
--- a/providers/edge3/src/airflow/providers/edge3/cli/worker.py
+++ b/providers/edge3/src/airflow/providers/edge3/cli/worker.py
@@ -388,6 +388,7 @@ class EdgeWorker:
server=self._execution_api_server_url,
log_path=workload.log_path,
)
+ results_queue.put("OK")
return 0
except Exception as e:
logger.exception("Task execution failed")
@@ -540,30 +541,35 @@ class EdgeWorker:
self.background_tasks.add(task)
task.add_done_callback(self.background_tasks.discard)
- while job.is_running:
+ while job.is_running and results_queue.empty():
await self._push_logs_in_chunks(job)
for _ in range(0, self.job_poll_interval * 10):
await sleep(0.1)
if not job.is_running:
break
await self._push_logs_in_chunks(job)
+ supervisor_msg = (
+ "(Unknown error, no exception details available)"
+ if results_queue.empty()
+ else results_queue.get()
+ )
+ # Ensure that supervisor really ended after we grabbed results from
queue
+ while job.is_running:
+ await sleep(0.1)
self.jobs.remove(job)
if job.is_success:
logger.info("Job completed: %s", job.edge_job.identifier)
await jobs_set_state(job.edge_job.key, TaskInstanceState.SUCCESS)
else:
- if results_queue.empty():
- ex_txt = "(Unknown error, no exception details available)"
- else:
- ex = results_queue.get()
- ex_txt = "\n".join(traceback.format_exception(ex))
- logger.error("Job failed: %s with:\n%s", job.edge_job.identifier,
ex_txt)
+ if isinstance(supervisor_msg, Exception):
+ supervisor_msg =
"\n".join(traceback.format_exception(supervisor_msg))
+ logger.error("Job failed: %s with:\n%s", job.edge_job.identifier,
supervisor_msg)
# Push it upwards to logs for better diagnostic as well
await logs_push(
task=job.edge_job.key,
log_chunk_time=timezone.utcnow(),
- log_chunk_data=f"Error executing job:\n{ex_txt}",
+ log_chunk_data=f"Error executing job:\n{supervisor_msg}",
)
await jobs_set_state(job.edge_job.key, TaskInstanceState.FAILED)
diff --git a/providers/edge3/tests/unit/edge3/cli/test_worker.py
b/providers/edge3/tests/unit/edge3/cli/test_worker.py
index be33ea68d36..d056d3af1e1 100644
--- a/providers/edge3/tests/unit/edge3/cli/test_worker.py
+++ b/providers/edge3/tests/unit/edge3/cli/test_worker.py
@@ -16,6 +16,7 @@
# under the License.
from __future__ import annotations
+import asyncio
import contextlib
import importlib
import json
@@ -81,6 +82,7 @@ MOCK_COMMAND = {
class _MockProcess(Process):
def __init__(self, returncode=None):
self.generated_returncode = None
+ self._is_alive = False
def poll(self):
pass
@@ -89,6 +91,18 @@ class _MockProcess(Process):
def returncode(self):
return self.generated_returncode
+ def is_alive(self):
+ return self._is_alive
+
+ def terminate(self):
+ self._is_alive = False
+
+ def kill(self):
+ self._is_alive = False
+
+ def join(self, timeout=None):
+ pass
+
class TestEdgeWorker:
@pytest.fixture(autouse=True)
@@ -219,7 +233,7 @@ class TestEdgeWorker:
result = worker_with_job._run_job_via_supervisor(edge_job.command, q)
assert result == 0
- q.put.assert_not_called()
+ q.put.assert_called_once()
@patch("airflow.sdk.execution_time.supervisor.supervise")
@pytest.mark.asyncio
@@ -296,6 +310,60 @@ class TestEdgeWorker:
assert len(worker_with_job.jobs) == 1 # no new job added (was removed
at the end...)
mock_logs_push.assert_not_called()
+ @patch("airflow.sdk.execution_time.supervisor.supervise")
+ @patch("airflow.providers.edge3.cli.worker.jobs_fetch")
+ @patch("airflow.providers.edge3.cli.worker.jobs_set_state")
+
@patch("airflow.providers.edge3.cli.worker.EdgeWorker._push_logs_in_chunks")
+ @patch("airflow.providers.edge3.cli.worker.logs_push")
+ @pytest.mark.asyncio
+ async def test_fetch_and_run_job_possible_deadlock(
+ self,
+ mock_logs_push,
+ mock_push_log_chunks,
+ mock_jobs_set_state,
+ mock_jobs_fetch,
+ mock_supervise,
+ worker_with_job: EdgeWorker,
+ ):
+ """Verify that a large exception from the subprocess does not deadlock
fetch_and_run_job."""
+
+ large_exception = Exception(f"Task execution failed with large error
message {'-' * 66000}")
+ mock_supervise.side_effect = large_exception
+
+ mock_jobs_fetch.side_effect = [
+ EdgeJobFetched(
+ dag_id="test",
+ task_id="test",
+ run_id="test",
+ map_index=-1,
+ try_number=1,
+ concurrency_slots=1,
+ command=MOCK_COMMAND, # type: ignore[arg-type]
+ ),
+ None,
+ ]
+ worker_with_job.__dict__["_execution_api_server_url"] =
"https://mock-server/execution";
+ worker_with_job.concurrency = 1 # only one job at a time
+ assert worker_with_job.free_concurrency == 0
+
+ try:
+ await asyncio.wait_for(worker_with_job.fetch_and_run_job(),
timeout=10.0)
+ except asyncio.TimeoutError:
+ # Clean up any hanging subprocess to prevent blocking pytest
+ for job in list(worker_with_job.jobs):
+ if job.process.is_alive():
+ job.process.terminate()
+ job.process.join(timeout=1.0)
+ if job.process.is_alive():
+ job.process.kill()
+ job.process.join()
+ pytest.fail("fetch_and_run_job timed out after 10s - DEADLOCK
DETECTED. ")
+
+ # If we reach here without timeout, the deadlock was not triggered
+ assert mock_jobs_set_state.call_count >= 1
+ mock_push_log_chunks.assert_called()
+ assert len(worker_with_job.jobs) <= 1 # new job removed, original
fixture job still there
+
@patch("airflow.providers.edge3.cli.worker.jobs_fetch")
@patch("airflow.providers.edge3.cli.worker.EdgeWorker._launch_job",
return_value=(Process(), Queue()))
@patch("airflow.providers.edge3.cli.worker.jobs_set_state")
