(airflow) branch main updated: Better error handling to prevent flaky test_last_chance_exception_handling test (#44909)

[ash]

This is an automated email from the ASF dual-hosted git repository.

ash 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 a56b03b9363 Better error handling to prevent flaky 
test_last_chance_exception_handling test (#44909)
a56b03b9363 is described below

commit a56b03b93634438e656b349be22298691e816666
Author: Ash Berlin-Taylor <a...@apache.org>
AuthorDate: Fri Dec 13 11:25:11 2024 +0000

    Better error handling to prevent flaky test_last_chance_exception_handling 
test (#44909)
    
    If the launched subprocess (which in tests just _immediately_ raises an
    Exception) exits very quickly before we even try to send the startup message
    it would fail with a BrokenPipeError. All we need to do in this case is 
handle
    it as the exit code of the task and it's message (which we already test) 
will
    cover it.
    
    This particular behaviour is hard to reliably catch in tests, so no tests 
are
    added here.
---
 task_sdk/src/airflow/sdk/execution_time/supervisor.py | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/task_sdk/src/airflow/sdk/execution_time/supervisor.py 
b/task_sdk/src/airflow/sdk/execution_time/supervisor.py
index cb5554681b3..677030b7bdc 100644
--- a/task_sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task_sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -412,8 +412,13 @@ class WatchedSubprocess:
 
         # Send the message to tell the process what it needs to execute
         log.debug("Sending", msg=msg)
-        self.stdin.write(msg.model_dump_json().encode())
-        self.stdin.write(b"\n")
+
+        try:
+            self.stdin.write(msg.model_dump_json().encode())
+            self.stdin.write(b"\n")
+        except BrokenPipeError:
+            # Debug is fine, the process will have shown _something_ in it's 
last_chance exception handler
+            log.debug("Couldn't send startup message to Subprocess - it died 
very early", pid=self.pid)
 
     def kill(
         self,