Fixed thread local _sentinel.callers defect and added test cases

[rahulgoyal2987]

---

^ 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.

[ashb]

Please add a meaningful description to this pr, and likely unit tests too

[rahulgoyal2987]

PFA

[rahulgoyal2987]

PFA

[dabla]

Is there a reason why this PR isn't merged yet as it fixes a thread safety issues with the ExecutorSafeguard?

[potiuk]

I thi k we mostly overlooked it :(

[kulkarni-sp]

What is the target release version for this fix?

[potiuk]

Currently Airflow 3. Why? Is this worthwhile to make an attempt to backport it to Airflow 2.* ? If so - why? What would be your arguments @kulkarni-sp ?

[kulkarni-sp]

Currently Airflow 3. Why? Is this worthwhile to make an attempt to backport it to Airflow 2.* ? If so - why? What would be your arguments @kulkarni-sp ?

We are currently using Airflow 2.9.3 and encountered the same errors during our attempt to upgrade to 2.10.3. Therefore, this fix is essential to unblock our upgrade path.

[potiuk]

We are currently using Airflow 2.9.3 and encountered the same errors during our attempt to upgrade to 2.10.3. Therefore, this fix is essential to unblock our upgrade path.

Could you please be more specific - what exactly error you experienced - i think that one has no clear issue that it marks as "solving" ? And is it possible that you apply that patch to verify that this one solves it ?

[kulkarni-sp]

We are currently using Airflow 2.9.3 and encountered the same errors during our attempt to upgrade to 2.10.3. Therefore, this fix is essential to unblock our upgrade path.

Could you please be more specific - what exactly error you experienced - i think that one has no clear issue that it marks as "solving" ? And is it possible that you apply that patch to verify that this one solves it ?

We are utilizing Airflow with KubernetesExecutor and have several long-running pods that continuously monitor specified locations for input files. Initially, we used mapped tasks to specify different input locations, but this led to a higher number of pods and caused stability issues on our AKS cluster. Now, we are employing ThreadPoolExecutor for parallel processing, which triggers the processor DAG upon receiving a valid input file. However, with Airflow 2.10.3, we encounter the following errors when triggering the processor DAG:

[2025-01-10, 12:53:29 UTC] {ThreadPoolExecutor-1_3 logging_mixin.py:190} INFO - pa: Error while calling processor DAG: '_thread._local' object has no attribute 'callers'

@task(executor_config=k8s_poller_exec_config_resource_requirements)
    def poll_adls_driver(config_list, **kwargs):
        args_list = []

        for upstream_config in config_list:
            args_list.append((upstream_config, kwargs,))

        with ThreadPoolExecutor(max_workers=5) as executor:
            # Unpack the argument tuples
            futures = [executor.submit(poll_adls, *args) for args in args_list]
        for future in futures:
            future.result()

[kulkarni-sp]

@potiuk We have also applied this patch, Requires minor change for it to work in multithreaded env i.e. getattr check and initialization should be done before sentinel check OR It could work with this PR https://github.com/apache/airflow/pull/44240/files

        sentinel_key = f"{self.__class__.__name__}__sentinel"
        sentinel = kwargs.pop(sentinel_key, None)
        //Initialize attribute callers
         if not getattr(cls._sentinel, "callers", None):
            cls._sentinel.callers = {}
        if sentinel:                
            cls._sentinel.callers[sentinel_key] = sentinel
        else:                
            sentinel = cls._sentinel.callers.pop(f"{func.__qualname__.split('.')[0]}__sentinel", None)
        ----

[eladkal]

I guess this issue fixes #44648 which is a bug on 2.10.3
@utkarsharma2 can we backport the fix to v2-10 branch?

[potiuk]

I marked it as 2.10.5 milestone to not forget about it.

[kulkarni-sp]

@utkarsharma2 Is it possible for you to include this make minor change? Essentially doing getattr check before if statement.
//Initialize attribute callers
if not getattr(cls._sentinel, "callers", None):
cls._sentinel.callers = {}
if sentinel:
cls._sentinel.callers[sentinel_key] = sentinel
else:
sentinel = cls._sentinel.callers.pop(f"{func.qualname.split('.')[0]}__sentinel", None)
----