jscheffl commented on PR #65838: URL: https://github.com/apache/airflow/pull/65838#issuecomment-4320110786
In general it is working as expected - the edge worker in my test case (integration test Dag witht he long running task) kills the task after my timeout. But the results are really a bit hard killing. In the task execution logs I see: <img width="488" height="71" alt="image" src="https://github.com/user-attachments/assets/4d4fdf3b-bc8b-499b-8cc9-58759bfac8e0"; /> In the CLI/Console of the execution I have: ``` 2026-04-25T16:50:38.366855Z [info ] 1 job is running [airflow.providers.edge3.cli.worker] loc=worker.py:472 2026-04-25T16:50:38.367040Z [warning ] Drain timeout of 30s exceeded with 1 job(s) still running. Sending SIGTERM. [airflow.providers.edge3.cli.worker] loc=worker.py:274 2026-04-25T16:50:38.487873Z [error ] Job failed: dag_id=integration_test task_id=long_running run_id=manual__2026-04-25T16:42:33.575750+00:00 map_index=-1 try_number=2 with: (Unknown error, no exception details available) [airflow.providers.edge3.cli.worker] loc=worker.py:525 2026-04-25T16:50:38.567889Z [info ] Quitting worker, signal being offline. [airflow.providers.edge3.cli.worker] loc=worker.py:429 [Breeze:3.12.13] root@c482c103c987:/opt/airflow$ --- Supervised process Last chance exception handler --- Traceback (most recent call last): File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1283, in run result = _execute_task(context=context, ti=ti, log=log) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1698, in _execute_task result = ctx.run(execute, context=context) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 443, in wrapper return func(self, *args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/task-sdk/src/airflow/sdk/bases/decorator.py", line 398, in execute return_value = super().execute(context) ^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 443, in wrapper return func(self, *args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py", line 227, in execute return_value = self.execute_callable() ^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py", line 250, in execute_callable return runner.run(*self.op_args, **self.op_kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/callback_runner.py", line 97, in run return func(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^ File "/files/dags/integration_test.py", line 175, in long_running print(f"Running for {i + 1} minutes now.") BrokenPipeError: [Errno 32] Broken pipe During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1367, in run log.exception("Task failed with exception") File "/usr/python/lib/python3.12/site-packages/structlog/_native.py", line 47, in exception return self.error(event, *args, **kw) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/task-sdk/src/airflow/sdk/_shared/logging/structlog.py", line 96, in meth return self._proxy_to_logger(name, event, **kw) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/python/lib/python3.12/site-packages/structlog/_base.py", line 223, in _proxy_to_logger return getattr(self._logger, method_name)(*args, **kw) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/python/lib/python3.12/site-packages/structlog/_output.py", line 321, in msg self._write(message + b"\n") BrokenPipeError: [Errno 32] Broken pipe During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1898, in main state, _, error = run(ti, context, log) ^^^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1372, in run f"ti.finish.{ti.dag_id}.{ti.task_id}.{state.value}", ^^^^^ UnboundLocalError: cannot access local variable 'state' where it is not associated with a value During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/supervisor.py", line 425, in _fork_main target() File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/supervisor.py", line 218, in _subprocess_main main() File "/usr/python/lib/python3.12/contextlib.py", line 81, in inner return func(*args, **kwds) ^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1905, in main log.exception("Top level error") File "/usr/python/lib/python3.12/site-packages/structlog/_native.py", line 47, in exception return self.error(event, *args, **kw) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/task-sdk/src/airflow/sdk/_shared/logging/structlog.py", line 96, in meth return self._proxy_to_logger(name, event, **kw) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/python/lib/python3.12/site-packages/structlog/_base.py", line 223, in _proxy_to_logger return getattr(self._logger, method_name)(*args, **kw) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/python/lib/python3.12/site-packages/structlog/_output.py", line 321, in msg self._write(message + b"\n") BrokenPipeError: [Errno 32] Broken pipe ``` Would it be possible to have a better controlled shutdown like an external kill being signalled such that the task logs have a clear signal what was the reason? Else this looks like an infrastructure error and you'd have a long time to research what the root cause is. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
