Sensor returns success state when fatal executor exception

[awdavidson]

Apache Airflow version

Other Airflow 2 version (please specify below)

If "Other Airflow 2 version" selected, which one?

2.9.3

What happened?

Execution of a sensor that pokes every 5 minutes failed within a celery worker with the following error message:

[2024-10-18 17:11:41,311: INFO/MainProcess] Task airflow.providers.celery.executors.celery_executor_utils.execute_command[73e87a6c-2aba-4982-ba21-80cf5dfb5bcd] received
[2024-10-18 17:11:41,327: INFO/ForkPoolWorker-2] [73e87a6c-2aba-4982-ba21-80cf5dfb5bcd] Executing command in Celery: ['airflow', 'tasks', 'run', 'IncSims_TESTTWIN', 'US.WaitForTap', 'scheduled__2024-10-18T10:15:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/test_dag.py']
[2024-10-18 17:18:41,836: ERROR/ForkPoolWorker-2] [73e87a6c-2aba-4982-ba21-80cf5dfb5bcd] Failed to execute task.
Traceback (most recent call last):  File "/home/airflow/.local/lib/python3.10/site-packages/celery/app/trace.py", line 453, in trace_task    R = retval = fun(*args, **kwargs)  File "/home/airflow/.local/lib/python3.10/site-packages/celery/app/trace.py", line 736, in __protected_call__    return self.run(*args, **kwargs)  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/providers/celery/executors/celery_executor_utils.py", line 136, in execute_command    _execute_in_fork(command_to_exec, celery_task_id)  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/providers/celery/executors/celery_executor_utils.py", line 151, in _execute_in_fork    raise AirflowException(msg)airflow.exceptions.AirflowException: Celery command failed on host: worker.host.com with celery_task_id 73e87a6c-2aba-4982-ba21-80cf5dfb5bcd (PID: 24011, Return Code: 256)
[2024-10-18 17:18:43,196: ERROR/ForkPoolWorker-2] Task airflow.providers.celery.executors.celery_executor_utils.execute_command[73e87a6c-2aba-4982-ba21-80cf5dfb5bcd] raised unexpected: AirflowException('Celery command failed on host: worker.host.com with celery_task_id 73e87a6c-2aba-4982-ba21-80cf5dfb5bcd (PID: 24011, Return Code: 256)')

The scheduler errors with

[2024-10-18 17:19:03 UTC] {scheduler_job_runner.py:846} ERROR - The executor reported that the task instance <TaskInstance: [someTaskInstance](http://sometaskinstance/).test scheduled__2024-10-18T16:00:00+00:00 [queued]> finished with state success, but the task instance's state attribute is queued. Learn more: https://airflow.apache.org/docs/apache-airflow/stable/troubleshooting.html#task-state-changed-externally

As a result, the task was marked as success and downstream tasks were executed.

What you think should happen instead?

The sensor's conditional was not true so running downstream tasks incorrectly run.

If the execution of the task failed, I would expect airflow to mark the task instance as failed and not success.

How to reproduce

Unfortunately, I haven't found an easy way to reproduce the issue

Operating System

Debian GNU/Linux 12 (bookworm)

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

Running on kubernetes cluster using celery executors with redis.

Using a postgres database for metadata

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[potiuk]

I think this is a low-level failure - becuase there was not enough memory for example (because apparently just starting a new process faile. When you run out of memory you cannot do much other than failing things - and likely airflow will not be able to do anything .. because it will have no memory to do anything

But it's hard to say - since it is not reproducible, and no obvious solution, we can at most convert it into discussion and maybe someone will have similar issue and more details/reproducible case.

[awdavidson]

With a low level failure I would still expect airflow to report the task instance with state failed rather than success?