LocalExecutor crashes in KinD Deployment with some Dags

[jscheffl]

Body

Attempted to enable K8s tests with LocalExecutor in #47472 and failed in making LocalExecutor working.

Branch: latest main on 2025-03-07

How to re-produce:
Use breeze k8s commands to set-up a KinD cluster, deploy airflow.
Python 3.9, Postgres, LocalExecutor

Open a shell to K8s cluster and airflow dags trigger example_bash_operator + airflow dags unpause example_bash_operator.

Check logs for scheduler Pod (breeze k8s k9s is your friend)

Shortcut if #47472 is merged: breeze k8s run-complete-tests --rebuild-base-image --python 3.9 --executor LocalExecutor - see tests in kubernetes_tests/test_other_executors.py - and these should NOT xfail!

When fixing please remove the XFail in kubernetes_tests/test_other_executors.py in the fix PR.

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.

[jscheffl]

Error trace 1 I could capture from scheduler POD stdout:

[2025-03-07T20:54:18.258+0000] {_client.py:1026} INFO - HTTP Request: PATCH http://airflow-api-server:8080/execution/task-instances/01957263-27fd-78fb-8e16-57907c484768/st
[2025-03-07T20:54:18.259+0000] {_trace.py:47} DEBUG - receive_response_body.started request=<Request [b'PATCH']>                                                           
[2025-03-07T20:54:18.259+0000] {_trace.py:47} DEBUG - receive_response_body.complete                                                                                       
[2025-03-07T20:54:18.259+0000] {_trace.py:47} DEBUG - response_closed.started                                                                                              
[2025-03-07T20:54:18.259+0000] {_trace.py:47} DEBUG - response_closed.complete                                                                                             
2025-03-07 20:54:18 [debug    ] Workload process exited        [supervisor] exit_code=0                                                                                    
2025-03-07 20:54:18 [info     ] Task finished                  [supervisor] duration=1.4302356929983944 exit_code=0 final_state=success                                    
[2025-03-07 20:59:21 +0000] [62] [INFO] Worker exiting (pid: 62)                                                                                                           
[2025-03-07 20:59:21 +0000] [63] [INFO] Worker exiting (pid: 63)                                                                                                           
[2025-03-07T20:59:21.527+0000] {scheduler_job_runner.py:232} INFO - Exiting gracefully upon receiving signal 15                                                            
[2025-03-07 20:59:21 +0000] [61] [INFO] Handling signal: term                                                                                                              
[2025-03-07T20:59:21.527+0000] {listener.py:37} DEBUG - Calling 'before_stopping' with {'component': <airflow.jobs.job.Job object at 0x7ef6b162d820>}                      
[2025-03-07T20:59:21.528+0000] {listener.py:38} DEBUG - Hook impls: []                                                                                                     
[2025-03-07T20:59:21.528+0000] {listener.py:42} DEBUG - Result from 'before_stopping': []                                                                                  
[2025-03-07T20:59:21.535+0000] {cli_action_loggers.py:97} DEBUG - Calling callbacks: []                                                                                    
[2025-03-07 20:59:21 +0000] [61] [INFO] Shutting down: Master                                                                                                              
stream closed EOF for airflow/airflow-scheduler-0 (scheduler)     

After this the POD was dead, unfortunately KinD/Liveness probe did not restart scheduler :-(

[jscheffl]

Error trace 2 I could capture from scheduler POD stdout:

2025-03-07 21:14:29 [debug    ] Workload process exited        [supervisor] exit_code=<Negsignal.SIGKILL: -9>                                                              
2025-03-07 21:14:29 [info     ] Process exited                 [supervisor] exit_code=<Negsignal.SIGKILL: -9> pid=153 signal=SIGKILL                                       
[2025-03-07T21:14:29.228+0000] {local_executor.py:101} ERROR - uhoh                                                                                                        
Traceback (most recent call last):                                                                                                                                         
  File "/opt/airflow/airflow/executors/local_executor.py", line 97, in _run_worker                                                                                         
    _execute_work(log, workload)                                                                                                                                           
  File "/opt/airflow/airflow/executors/local_executor.py", line 118, in _execute_work                                                                                      
    supervise(                                                                                                                                                             
  File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/supervisor.py", line 1067, in supervise                                                                       
    process = ActivitySubprocess.start(                                                                                                                                    
  File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/supervisor.py", line 665, in start                                                                            
    proc._on_child_started(ti=what, dag_rel_path=dag_rel_path, bundle_info=bundle_info)                                                                                    
  File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/supervisor.py", line 675, in _on_child_started                                                                
    ti_context = self.client.task_instances.start(ti.id, self.pid, start_date)                                                                                             
  File "/opt/airflow/task-sdk/src/airflow/sdk/api/client.py", line 134, in start                                                                                           
    resp = self.client.patch(f"task-instances/{id}/run", content=body.model_dump_json())                                                                                   
  File "/home/airflow/.local/lib/python3.9/site-packages/httpx/_client.py", line 1219, in patch                                                                            
    return self.request(                                                                                                                                                   
  File "/home/airflow/.local/lib/python3.9/site-packages/tenacity/__init__.py", line 336, in wrapped_f                                                                     
    return copy(f, *args, **kw)                                                                                                                                            
  File "/home/airflow/.local/lib/python3.9/site-packages/tenacity/__init__.py", line 475, in __call__                                                                      
    do = self.iter(retry_state=retry_state)                                                                                                                                
  File "/home/airflow/.local/lib/python3.9/site-packages/tenacity/__init__.py", line 376, in iter                                                                          
    result = action(retry_state)                                                                                                                                           
  File "/home/airflow/.local/lib/python3.9/site-packages/tenacity/__init__.py", line 398, in <lambda>                                                                      
    self._add_action_func(lambda rs: rs.outcome.result())                                                                                                                  
  File "/usr/local/lib/python3.9/concurrent/futures/_base.py", line 439, in result                                                                                         
    return self.__get_result()                                                                                                                                             
  File "/usr/local/lib/python3.9/concurrent/futures/_base.py", line 391, in __get_result                                                                                   
    raise self._exception                                                                                                                                                  
  File "/home/airflow/.local/lib/python3.9/site-packages/tenacity/__init__.py", line 478, in __call__                                                                      
    result = fn(*args, **kwargs)                                                                                                                                           
  File "/opt/airflow/task-sdk/src/airflow/sdk/api/client.py", line 442, in request                                                                                         
    return super().request(*args, **kwargs)                                                                                                                                
  File "/home/airflow/.local/lib/python3.9/site-packages/httpx/_client.py", line 827, in request                                                                           
    return self.send(request, auth=auth, follow_redirects=follow_redirects)                                                                                                
  File "/home/airflow/.local/lib/python3.9/site-packages/httpx/_client.py", line 914, in send                                                                              
    response = self._send_handling_auth(                                                                                                                                   
  File "/home/airflow/.local/lib/python3.9/site-packages/httpx/_client.py", line 942, in _send_handling_auth                                                               
    response = self._send_handling_redirects(                                                                                                                              
  File "/home/airflow/.local/lib/python3.9/site-packages/httpx/_client.py", line 999, in _send_handling_redirects                                                          
    raise exc                                                                                                                                                              
  File "/home/airflow/.local/lib/python3.9/site-packages/httpx/_client.py", line 982, in _send_handling_redirects                                                          
    hook(response)                                                                                                                                                         
  File "/opt/airflow/task-sdk/src/airflow/sdk/api/client.py", line 99, in raise_on_4xx_5xx                                                                                 
    return get_json_error(response) or response.raise_for_status()                                                                                                         
  File "/opt/airflow/task-sdk/src/airflow/sdk/api/client.py", line 95, in get_json_error                                                                                   
    raise err                                                                                                                                                              
airflow.sdk.api.client.ServerResponseError: Server returned error                                    

...after this the scheduler POD was just hanging and un-responsive.