fix: ensure execute_command is registered with Celery app at worker s…

[kalluripradeep]

When the Celery worker starts, it imports app from celery_executor.py but celery_executor_utils.py was never imported at module level. This means execute_command never gets registered as a Celery task when the worker boots, and the celery_import_modules signal handler never connects.

This worked before because older Celery versions were more lenient about when tasks got registered. Celery provider 3.17.0 exposed this — workers start receiving messages before _process_tasks() is ever called, so execute_command is unknown and messages get discarded:
KeyError: 'execute_command'
Received unregistered task of type 'execute_command'. The message has been ignored and discarded.

Fix is a single import in celery_executor.py so celery_executor_utils is loaded when the worker imports app at startup.

Fixes#63043

[potiuk]

Nice! Thanks for investigation!

[potiuk]

I added full_tests_needed to see if the prod image will be built and docker-compose tests run

[potiuk]

Nope. @kalluripradeep -> it does not seem to fix the problem.

[potiuk]

Looks like the problem is not fixed

[potiuk]

same error:

 The delivery info for this task is:
{'exchange': '', 'routing_key': 'default'}
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.10/site-packages/celery/worker/consumer/consumer.py", line 668, in on_task_received
strategy = strategies[type_]
KeyError: 'execute_command'

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

[potiuk]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

@potiuk — opened #63110 targeting main with the same fix plus a regression test. Module-level import of celery_executor_utils in celery_executor.py so execute_workload and execute_command are registered when the worker boots.

[kalluripradeep]

When the Celery worker starts, it imports app from celery_executor.py but celery_executor_utils.py was never imported at module level. This means execute_command never gets registered as a Celery task when the worker boots, and the celery_import_modules signal handler never connects.

This worked before because older Celery versions were more lenient about when tasks got registered. Celery provider 3.17.0 exposed this — workers start receiving messages before _process_tasks() is ever called, so execute_command is unknown and messages get discarded:
KeyError: 'execute_command'
Received unregistered task of type 'execute_command'. The message has been ignored and discarded.

Fix is a single import in celery_executor.py so celery_executor_utils is loaded when the worker imports app at startup.

Fixes#63043

[potiuk]

Nice! Thanks for investigation!

[potiuk]

I added full_tests_needed to see if the prod image will be built and docker-compose tests run

[potiuk]

Nope. @kalluripradeep -> it does not seem to fix the problem.

[potiuk]

Looks like the problem is not fixed

[potiuk]

same error:

 The delivery info for this task is:
{'exchange': '', 'routing_key': 'default'}
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.10/site-packages/celery/worker/consumer/consumer.py", line 668, in on_task_received
strategy = strategies[type_]
KeyError: 'execute_command'

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

[potiuk]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

@potiuk — opened #63110 targeting main with the same fix plus a regression test. Module-level import of celery_executor_utils in celery_executor.py so execute_workload and execute_command are registered when the worker boots.

[kalluripradeep]

When the Celery worker starts, it imports app from celery_executor.py but celery_executor_utils.py was never imported at module level. This means execute_command never gets registered as a Celery task when the worker boots, and the celery_import_modules signal handler never connects.

This worked before because older Celery versions were more lenient about when tasks got registered. Celery provider 3.17.0 exposed this — workers start receiving messages before _process_tasks() is ever called, so execute_command is unknown and messages get discarded:
KeyError: 'execute_command'
Received unregistered task of type 'execute_command'. The message has been ignored and discarded.

Fix is a single import in celery_executor.py so celery_executor_utils is loaded when the worker imports app at startup.

Fixes#63043

[potiuk]

Nice! Thanks for investigation!

[potiuk]

I added full_tests_needed to see if the prod image will be built and docker-compose tests run

[potiuk]

Nope. @kalluripradeep -> it does not seem to fix the problem.

[potiuk]

Looks like the problem is not fixed

[potiuk]

same error:

 The delivery info for this task is:
{'exchange': '', 'routing_key': 'default'}
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.10/site-packages/celery/worker/consumer/consumer.py", line 668, in on_task_received
strategy = strategies[type_]
KeyError: 'execute_command'

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

[potiuk]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

@potiuk — opened #63110 targeting main with the same fix plus a regression test. Module-level import of celery_executor_utils in celery_executor.py so execute_workload and execute_command are registered when the worker boots.

[kalluripradeep]

When the Celery worker starts, it imports app from celery_executor.py but celery_executor_utils.py was never imported at module level. This means execute_command never gets registered as a Celery task when the worker boots, and the celery_import_modules signal handler never connects.

This worked before because older Celery versions were more lenient about when tasks got registered. Celery provider 3.17.0 exposed this — workers start receiving messages before _process_tasks() is ever called, so execute_command is unknown and messages get discarded:
KeyError: 'execute_command'
Received unregistered task of type 'execute_command'. The message has been ignored and discarded.

Fix is a single import in celery_executor.py so celery_executor_utils is loaded when the worker imports app at startup.

Fixes#63043

[potiuk]

Nice! Thanks for investigation!

[potiuk]

I added full_tests_needed to see if the prod image will be built and docker-compose tests run

[potiuk]

Nope. @kalluripradeep -> it does not seem to fix the problem.

[potiuk]

Looks like the problem is not fixed

[potiuk]

same error:

 The delivery info for this task is:
{'exchange': '', 'routing_key': 'default'}
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.10/site-packages/celery/worker/consumer/consumer.py", line 668, in on_task_received
strategy = strategies[type_]
KeyError: 'execute_command'

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

[potiuk]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

@potiuk — opened #63110 targeting main with the same fix plus a regression test. Module-level import of celery_executor_utils in celery_executor.py so execute_workload and execute_command are registered when the worker boots.

[kalluripradeep]

When the Celery worker starts, it imports app from celery_executor.py but celery_executor_utils.py was never imported at module level. This means execute_command never gets registered as a Celery task when the worker boots, and the celery_import_modules signal handler never connects.

This worked before because older Celery versions were more lenient about when tasks got registered. Celery provider 3.17.0 exposed this — workers start receiving messages before _process_tasks() is ever called, so execute_command is unknown and messages get discarded:
KeyError: 'execute_command'
Received unregistered task of type 'execute_command'. The message has been ignored and discarded.

Fix is a single import in celery_executor.py so celery_executor_utils is loaded when the worker imports app at startup.

Fixes#63043

[potiuk]

Nice! Thanks for investigation!

[potiuk]

I added full_tests_needed to see if the prod image will be built and docker-compose tests run

[potiuk]

Nope. @kalluripradeep -> it does not seem to fix the problem.

[potiuk]

Looks like the problem is not fixed

[potiuk]

same error:

 The delivery info for this task is:
{'exchange': '', 'routing_key': 'default'}
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.10/site-packages/celery/worker/consumer/consumer.py", line 668, in on_task_received
strategy = strategies[type_]
KeyError: 'execute_command'

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

[potiuk]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

@potiuk — opened #63110 targeting main with the same fix plus a regression test. Module-level import of celery_executor_utils in celery_executor.py so execute_workload and execute_command are registered when the worker boots.

[kalluripradeep]

When the Celery worker starts, it imports app from celery_executor.py but celery_executor_utils.py was never imported at module level. This means execute_command never gets registered as a Celery task when the worker boots, and the celery_import_modules signal handler never connects.

This worked before because older Celery versions were more lenient about when tasks got registered. Celery provider 3.17.0 exposed this — workers start receiving messages before _process_tasks() is ever called, so execute_command is unknown and messages get discarded:
KeyError: 'execute_command'
Received unregistered task of type 'execute_command'. The message has been ignored and discarded.

Fix is a single import in celery_executor.py so celery_executor_utils is loaded when the worker imports app at startup.

Fixes#63043

[potiuk]

Nice! Thanks for investigation!

[potiuk]

I added full_tests_needed to see if the prod image will be built and docker-compose tests run

[potiuk]

Nope. @kalluripradeep -> it does not seem to fix the problem.

[potiuk]

Looks like the problem is not fixed

[potiuk]

same error:

 The delivery info for this task is:
{'exchange': '', 'routing_key': 'default'}
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.10/site-packages/celery/worker/consumer/consumer.py", line 668, in on_task_received
strategy = strategies[type_]
KeyError: 'execute_command'

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

[potiuk]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

@potiuk — opened #63110 targeting main with the same fix plus a regression test. Module-level import of celery_executor_utils in celery_executor.py so execute_workload and execute_command are registered when the worker boots.

[kalluripradeep]

When the Celery worker starts, it imports app from celery_executor.py but celery_executor_utils.py was never imported at module level. This means execute_command never gets registered as a Celery task when the worker boots, and the celery_import_modules signal handler never connects.

This worked before because older Celery versions were more lenient about when tasks got registered. Celery provider 3.17.0 exposed this — workers start receiving messages before _process_tasks() is ever called, so execute_command is unknown and messages get discarded:
KeyError: 'execute_command'
Received unregistered task of type 'execute_command'. The message has been ignored and discarded.

Fix is a single import in celery_executor.py so celery_executor_utils is loaded when the worker imports app at startup.

Fixes#63043

[potiuk]

Nice! Thanks for investigation!

[potiuk]

I added full_tests_needed to see if the prod image will be built and docker-compose tests run

[potiuk]

Nope. @kalluripradeep -> it does not seem to fix the problem.

[potiuk]

Looks like the problem is not fixed

[potiuk]

same error:

 The delivery info for this task is:
{'exchange': '', 'routing_key': 'default'}
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.10/site-packages/celery/worker/consumer/consumer.py", line 668, in on_task_received
strategy = strategies[type_]
KeyError: 'execute_command'

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

[potiuk]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

@potiuk — opened #63110 targeting main with the same fix plus a regression test. Module-level import of celery_executor_utils in celery_executor.py so execute_workload and execute_command are registered when the worker boots.

[kalluripradeep]

When the Celery worker starts, it imports app from celery_executor.py but celery_executor_utils.py was never imported at module level. This means execute_command never gets registered as a Celery task when the worker boots, and the celery_import_modules signal handler never connects.

This worked before because older Celery versions were more lenient about when tasks got registered. Celery provider 3.17.0 exposed this — workers start receiving messages before _process_tasks() is ever called, so execute_command is unknown and messages get discarded:
KeyError: 'execute_command'
Received unregistered task of type 'execute_command'. The message has been ignored and discarded.

Fix is a single import in celery_executor.py so celery_executor_utils is loaded when the worker imports app at startup.

Fixes#63043

[potiuk]

Nice! Thanks for investigation!

[potiuk]

I added full_tests_needed to see if the prod image will be built and docker-compose tests run

[potiuk]

Nope. @kalluripradeep -> it does not seem to fix the problem.

[potiuk]

Looks like the problem is not fixed

[potiuk]

same error:

 The delivery info for this task is:
{'exchange': '', 'routing_key': 'default'}
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.10/site-packages/celery/worker/consumer/consumer.py", line 668, in on_task_received
strategy = strategies[type_]
KeyError: 'execute_command'

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

[potiuk]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

[kalluripradeep]

@potiuk — the fix is correct but the docker-compose test installs apache-airflow-providers-celery==3.17.0 from PyPI on top of the base image, so it doesn't pick up the change from this branch. The fix needs to be in the provider package and released as a new version. Should this PR target the provider package source instead, so it gets included in 3.17.1?

Yes. Initially the cahnge was to "airflow-core" - but since you changed it now to providers, it needs to be done for main and indeed next versioin of celery provider. Which is what I kind of expected. Closing it - open a new one for main and in the meantime #63046 should be merged to release 2.11.2

@potiuk — opened #63110 targeting main with the same fix plus a regression test. Module-level import of celery_executor_utils in celery_executor.py so execute_workload and execute_command are registered when the worker boots.