celery_executor got an unexpected keyword argument 'remove_running'

[potiuk]

Discussed in #39980

^{Originally posted by weitang2022 May 31, 2024}
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)
```</div>

[potiuk]

Hey @vincbeck @o-nikolas -> looks like we got it now when 2.9.2rc1 is using the old celery provider.

Way to reproduce:

1. Run this:

breeze start-airflow --use-airflow-version 2.9.2rc1 --load-example-dags --load-default-connections --executor CeleryExecutor --airflow-extras celery

2. Check installed packages:

root@c45830039533:/opt/airflow# pip freeze | grep celery
apache-airflow-providers-celery==3.7.1
celery==5.4.0
root@c45830039533:/opt/airflow# pip freeze | grep apache-airflow
apache-airflow==2.9.2rc1
apache-airflow-providers-celery==3.7.1
apache-airflow-providers-common-io==1.3.2
apache-airflow-providers-common-sql==1.14.0
apache-airflow-providers-fab==1.1.1
apache-airflow-providers-ftp==3.9.1
apache-airflow-providers-http==4.11.1
apache-airflow-providers-imap==3.6.1
apache-airflow-providers-smtp==1.7.1
apache-airflow-providers-sqlite==3.8.1
root@c45830039533:/opt/airflow#

3. Connect to http://localhost:28080/

4. Run "example_complex" dag

5. You will see BaseExecutor.change_state() got an unexpected keyword argument 'remove_running' airflow_scheduler - errors. I am afraid we will need to to handle back-compatibility there.

[o-nikolas]

#40035 (this one is just a unit test, but it links to the other issues and PR that fixes it as wel. That was merged a couple days ago)

[potiuk]

@o-nikolas -> but I think the problem is that the new executor interface does not work with old celery provider.

[potiuk]

(or the other way round - maybe the celery executor 3.7.1 does not work with 2.9.2 - and likely 2.8, 2.9 in general - not sure as I am on a meeting today).

[o-nikolas]

I'm AFK right now, I can look later, but the traceback seems to be base exec being called with a param it doesn't expect (and the type error is now caught correctly as of a couple days ago)

TypeError: BaseExecutor.change_state() got an unexpected keyword argument 'remove_running'

[potiuk]

Ok. I see those were not marked for 2.9.2 so they should be in this case.

[potiuk]

Marked them to be included in 2.9.2rc2

[vincbeck]

I marked it to be in 2.9.2 but it got removed :( But anyway, yes #40012 is the fix to this issue

[potiuk]

Actually - it's not a full fix. It will only work with "fufure" celery provider when it is released. The older versions of celery providers will not work with it @vincbeck @o-nikolas. I guess that means that we

1. have to wait with 2.9.2 until new celery provider is released
2. have to yank celery provider 3.7.1 because ECS Executor: Set tasks to RUNNING state once active #39212 essentially broke the executor for all past versions of Airflow, where Fixing exception types to include TypeError, which is what is raised in #40012 fixes it in the future celery provider

cc: @eladkal @ephraimbuddy

[potiuk]

This might actually justify slightly acceleratte celery /k8s provider voting, because we won't be able to effectively test 2.9.1rc1 until celery/k8s providers are released cc: @eladkal .

[eladkal]

1. Celery provider 3.7.2rc1 will be released before Airflow 2.9.2
2. Celery provider 3.7.1 is yanked

I guess we can close this issue?

[potiuk]

Yep.