Fix _SECRET and _CMD broker configuration

[JonnyWaffles]

Fixes _SECRET and _CMD broker configurations not working after 2.7 release

closes: #34612

Someone wiser than me may be able to come up with an effective test, but because the fix involves import time side effects I couldn't think of an additional unit/integration test that covers this edge case but can confirm that loading the providers before attempting to configure the celery settings via conf resolves the bug where _SECRET and _CMD cannot be used to configure the broker after 2.7 moved Celery into the providers package.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[hussein-awala]

Makes sense! Could you add a test for this change?

[JonnyWaffles]

Hey @hussein-awala I really don't know how to test this inside the kit because the root cause is import time side effects which are hard to recreate in a pytest context. If you or anyone else can assist me I would appreciate it, but its hard to recreate the bug outside of the actual context of running airflow celery worker with a broker_url_secret or a broker_url_cmd setting.

[JonnyWaffles]

Hi team just following up on this one. I think the existing test kit is sufficient, as I am not sure how to test the import time side effects this patch corrects. Any advice?

[uranusjr]

I wonder if we should put the entire app initialise logic into a function and have app = _get_celery_app() instead.

[potiuk]

I wonder if we should put the entire app initialise logic into a function and have app = _get_celery_app() instead.

Yes. This is better idea. Also @provider_configuration_loaded decorator should be used for that method rather than manual intialization.

[potiuk]

As @uranusjr proposed, can you extract _get_celery_app method instead. Also instead of running provider configuration manually, just decorate the method with @provider_configuration_loaded decorator similarly to worker celery command.

[potiuk]

BTW. I still do not understand how "worker" could be affected but at least this fix might fix the liveness probe issue mentioned in #34612 (comment). (which indeed might have the problem).

[hussein-awala]

I tested it and it looks good, I will try to add a test once this one is merged, to avoid breaking it in the future.

[JonnyWaffles]

It looks like the test kit uses the original singleton dictionary celery_executor_utils.celery_configuration a few times, so to reduce any headache I'm tempted to leave the original global and app instantiation alone, so the patch would be.

celery_configuration = None

@providers_configuration_loaded
def _get_celery_app(celery_app_name) -> Celery:
    """Init providers before importing the configuration, so the _SECRET and _CMD options work."""
    global celery_configuration

    if conf.has_option("celery", "celery_config_options"):
        celery_configuration = conf.getimport("celery", "celery_config_options")
    else:
        from airflow.providers.celery.executors.default_celery import DEFAULT_CELERY_CONFIG

        celery_configuration = DEFAULT_CELERY_CONFIG

        ...

    return Celery(celery_app_name, config_source=celery_configuration)
...

app = _get_celery_app(celery_app_name)

This way the global configuration is still available to those like the test kit that were manipulating it. Thoughts? @potiuk @uranusjr

[JonnyWaffles]

Apologies friends! I tried to rebase/merge via Git UI after your approvals. Not sure what other action I need to take. New to your process. Is there anything else I need to do to merge this PR? @hussein-awala / @potiuk

[potiuk]

Merged. thanks !

[eladkal]

Removing milestone as this is provider code