fix(providers/celery): Migrate conf imports to SDK compatibility layer

[ftakelait]

Summary

Migrates conf imports in the Celery provider from airflow.configuration to airflow.providers.common.compat.sdk for Airflow 3.x SDK compatibility.

Changes Made

Import Migration (9 files)

• celery_executor.py
• celery_executor_utils.py (2 imports)
• celery_kubernetes_executor.py
• celery_command.py
• default_celery.py
• Test files (4 files)

Fallback Parameters Added

Added fallback parameters to conf.get() calls that execute at module import time:

File	Config Key	Fallback
celery_executor.py	FLOWER_HOST	"0.0.0.0"
celery_executor.py	FLOWER_PORT	5555
celery_executor.py	FLOWER_URL_PREFIX	""
celery_executor.py	FLOWER_BASIC_AUTH	""
celery_executor.py	DEFAULT_QUEUE	"default"
celery_executor.py	worker_concurrency	16
celery_executor.py	SYNC_PARALLELISM	0
celery_executor.py	task_publish_max_retries	3
celery_executor_utils.py	operation_timeout	1.0
celery_executor_utils.py	CELERY_APP_NAME	"airflow.executors.celery_executor"
default_celery.py	SSL_ACTIVE	False

Why fallbacks are needed?

The SDK conf.get() is stricter than airflow.configuration.conf.get() - it raises AirflowConfigException when a key is not found instead of returning None. This was causing test collection failures when config keys were accessed at module import time before provider configuration was loaded.

Part of #60000

[sunank200]

Why do we need defaults?

[potiuk]

Still working on it @ftakelait ? are you going to rebase and fix the comments ?

[ftakelait]

Still working on it @ftakelait ? are you going to rebase and fix the comments ?

Hi @potiuk, I am still waiting for @sunank200 to reply about whether I should keep the fallback or I baypass the testing stage.

[potiuk]

I have not seen your question - it's not visible here - I don't think @sunank200 is rather waiting for you - looking at the comments in PR.

#60074 (comment) is where we discussed how to fix it - and hopefully it will be fixed there

But you can still rebase though and keep on solving conflicts. That would be my recommendation while waiting for mechanism implemented for k8s

[ftakelait]

Rebased the PR. I'll wait for the configuration mechanism being developed in #60074, then update this PR to use it instead of individual fallbacks.

[uranusjr]

There are unresolved conflict markers here.

[uranusjr]

Why do we need to add fallbacks to many configs? Did they not work without the fallbacks?

[potiuk]

Why do we need to add fallbacks to many configs? Did they not work without the fallbacks?

@uranusjr -> This is manifestation of doing too much and in a wrong sequence during imports. The problem is that currently the sdk config does not have the defaults "hard-coded" fallbacks for celery and kubernetes providers (@amoghrajesh deliberately skipped them when moving conf to task.sdk) - because initialization of executors (that already need those fallbacks) is happening during importing airflow and before provider's manager initializes configuration retrieved from providers. This has been discussed and I proposed to bring it back as a temporary solution before we finish task isolation and will be able to do explicit initialization rather than import-based initialization: #60074 (comment)

But maybe there are other proposals?