dheerajturaga commented on PR #65838: URL: https://github.com/apache/airflow/pull/65838#issuecomment-4321167204
> 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 alt="image" width="488" height="71" src="https://private-user-images.githubusercontent.com/95105677/583790198-4d4fdf3b-bc8b-499b-8cc9-58759bfac8e0.png?jwt=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3NzcxNzA5MzIsIm5iZiI6MTc3NzE3MDYzMiwicGF0aCI6Ii85NTEwNTY3Ny81ODM3OTAxOTgtNGQ0ZmRmM2ItYmM4Yi00OTliLThjYzktNTg3NTliZmFjOGUwLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNjA0MjYlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjYwNDI2VDAyMzAzMlomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTQwZGExYWQ3ODhkNWQ2MmEyNzkxN2RjNjc5Yzk1YjgwMjhhYWNjYWVmZmNkMDMzODIwNjY4OGVjZjYwMGQ2MzgmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JnJlc3BvbnNlLWNvbnRlbnQtdHlwZT1pbWFnZSUyRnBuZyJ9.gddYQVb4mVhDwOgfvLOYcenoKD87yBsP4GobTf1waag";> > > 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. @jscheffl this is task-sdk bug surfaced by any worker that SIGTERMs a task supervisor mid-run. When the supervisor's parent pipe closes, user-code print() raises BrokenPipeError which leads to another downstream BrokenPipes again from inside the except block. I now push a clear message into each running task's log stream, also push is best-effort (logs_push can fail mid-drain) <img width="1331" height="453" alt="image" src="https://github.com/user-attachments/assets/a15cd0a2-d7a7-459c-b98a-2e785a989c3d"; /> -- 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]
![https://private-user-images.githubusercontent.com/95105677/583790198-4d4fdf3b-bc8b-499b-8cc9-58759bfac8e0.png?jwt=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3NzcxNzA5MzIsIm5iZiI6MTc3NzE3MDYzMiwicGF0aCI6Ii85NTEwNTY3Ny81ODM3OTAxOTgtNGQ0ZmRmM2ItYmM4Yi00OTliLThjYzktNTg3NTliZmFjOGUwLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNjA0MjYlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjYwNDI2VDAyMzAzMlomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTQwZGExYWQ3ODhkNWQ2MmEyNzkxN2RjNjc5Yzk1YjgwMjhhYWNjYWVmZmNkMDMzODIwNjY4OGVjZjYwMGQ2MzgmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JnJlc3BvbnNlLWNvbnRlbnQtdHlwZT1pbWFnZSUyRnBuZyJ9.gddYQVb4mVhDwOgfvLOYcenoKD87yBsP4GobTf1waag"](https://private-user-images.githubusercontent.com/95105677/583790198-4d4fdf3b-bc8b-499b-8cc9-58759bfac8e0.png?jwt=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3NzcxNzA5MzIsIm5iZiI6MTc3NzE3MDYzMiwicGF0aCI6Ii85NTEwNTY3Ny81ODM3OTAxOTgtNGQ0ZmRmM2ItYmM4Yi00OTliLThjYzktNTg3NTliZmFjOGUwLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNjA0MjYlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjYwNDI2VDAyMzAzMlomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTQwZGExYWQ3ODhkNWQ2MmEyNzkxN2RjNjc5Yzk1YjgwMjhhYWNjYWVmZmNkMDMzODIwNjY4OGVjZjYwMGQ2MzgmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JnJlc3BvbnNlLWNvbnRlbnQtdHlwZT1pbWFnZSUyRnBuZyJ9.gddYQVb4mVhDwOgfvLOYcenoKD87yBsP4GobTf1waag"){kind=link}