celery_executor got an unexpected keyword argument 'remove_running'

[weitang2022]

I upgraded to Airflow 2.9.1. It's been running well for a week, but today I see the error below. I tried to downgrade to 2.8.3, but I still see the issue. All tasks are stuck in "scheduled" status. Any idea how to get around it? Thank you for your help!

May 31, 2024 at 13:51 (UTC-4:00) [�[34m2024-05-31T17:51:01.889+0000�[0m] {�[34mcelery_executor.py:�[0m370} ERROR�[0m - Error syncing the Celery executor, ignoring it.�[0m airflow_scheduler
May 31, 2024 at 13:51 (UTC-4:00) Traceback (most recent call last): airflow_scheduler
May 31, 2024 at 13:51 (UTC-4:00) File "/usr/local/lib/python3.11/site-packages/airflow/providers/celery/executors/celery_executor.py", line 362, in update_task_state airflow_scheduler
May 31, 2024 at 13:51 (UTC-4:00) self.success(key, info) airflow_scheduler
May 31, 2024 at 13:51 (UTC-4:00) File "/usr/local/lib/python3.11/site-packages/airflow/executors/base_executor.py", line 332, in success airflow_scheduler
May 31, 2024 at 13:51 (UTC-4:00) self.change_state(key, TaskInstanceState.SUCCESS, info) airflow_scheduler
May 31, 2024 at 13:51 (UTC-4:00) File "/usr/local/lib/python3.11/site-packages/airflow/providers/celery/executors/celery_executor.py", line 351, in change_state airflow_scheduler
May 31, 2024 at 13:51 (UTC-4:00) super().change_state(key, state, info, remove_running=remove_running) airflow_scheduler
May 31, 2024 at 13:51 (UTC-4:00) TypeError: BaseExecutor.change_state() got an unexpected keyword argument 'remove_running' airflow_scheduler

The code here is actually throwing a TypeError:

    def change_state(
        self, key: TaskInstanceKey, state: TaskInstanceState, info=None, remove_running=True
    ) -> None:
        try:
            super().change_state(key, state, info, remove_running=remove_running)
        except AttributeError:
            # Earlier versions of the BaseExecutor don't accept the remove_running parameter for this method
            # TODO: Remove when min airflow version >= 2.9.2
            super().change_state(key, state, info)
        self.tasks.pop(key, None)

[weitang2022]

Looks like apache-airflow-providers-celery had a new version released yesterday. Just downgrade it to 3.7.0 and it works again now.

[potiuk]

cc: @vincbeck @eladkal @ephraimbuddy -> it awfully looks like we have a backwards incompatible change in executor interface that should be fixed before we got to 2.9.2

[potiuk]

Not sure if I will have time to look at it in the next few days (getting to Community Over Code in Bratislava)

[o-nikolas]

We knew of the incompatibility but the exception we were trying to catch was the wrong one apparently 😭

I'll find where our test coverage missed and put out a fix in the next day or so.

[eladkal]

There is a pending PR with the fix #40012

[o-nikolas]

Ahhh, it's a python version thing, this makes a lot of sense how this snuck through.

[vincbeck]

@o-nikolas