Scheduler using KubernetesExecutor crashes when `sentry-sdk` is installed and enabled

[wolfier]

Apache Airflow version

2.11.0

If "Other Airflow 2 version" selected, which one?

No response

What happened?

The scheduler is failing to start. The errors seem to be originating from importing BaseOperator that is triggered by a custom plugin. This happens to other plugins too.

Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/airflow/plugins_manager.py", line 259, in load_entrypoint_plugins
    plugin_class = entry_point.load()
                   ^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/importlib/metadata/__init__.py", line 205, in load
    module = import_module(match.group('module'))
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/importlib/__init__.py", line 90, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<frozen importlib._bootstrap>", line 1387, in _gcd_import
  File "<frozen importlib._bootstrap>", line 1360, in _find_and_load
  File "<frozen importlib._bootstrap>", line 1331, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 935, in _load_unlocked
  File "<frozen importlib._bootstrap_external>", line 999, in exec_module
  File "<frozen importlib._bootstrap>", line 488, in _call_with_frames_removed
  File "/usr/local/lib/python3.12/site-packages/airflow/providers/openlineage/plugins/openlineage.py", line 21, in <module>
    from airflow.providers.openlineage.plugins.listener import get_openlineage_listener
  File "/usr/local/lib/python3.12/site-packages/airflow/providers/openlineage/plugins/listener.py", line 31, in <module>
    from airflow.models import DagRun, TaskInstance
  File "/usr/local/lib/python3.12/site-packages/airflow/models/__init__.py", line 79, in __getattr__
    val = import_string(f"{path}.{name}")
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/utils/module_loading.py", line 39, in import_string
    module = import_module(module_path)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/importlib/__init__.py", line 90, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/models/dagrun.py", line 60, in <module>
    from airflow.models.taskinstance import TaskInstance as TI
  File "/usr/local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 106, in <module>
    from airflow.sentry import Sentry
  File "/usr/local/lib/python3.12/site-packages/airflow/sentry.py", line 196, in <module>
    Sentry = ConfiguredSentry()
             ^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/sentry.py", line 88, in __init__
    executor_class, _ = ExecutorLoader.import_default_executor_cls(validate=False)
                        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/executors/executor_loader.py", line 314, in import_default_executor_cls
    executor, source = cls.import_executor_cls(executor_name, validate=validate)
                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/executors/executor_loader.py", line 302, in import_executor_cls
    return _import_and_validate(executor_name.module_path), executor_name.connector_source
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/executors/executor_loader.py", line 286, in _import_and_validate
    executor = import_string(path)
               ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/utils/module_loading.py", line 39, in import_string
    module = import_module(module_path)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/importlib/__init__.py", line 90, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/astronomer/kubernetes/executors/kubernetes_executor.py", line 39, in <module>
    from astronomer.kubernetes.pod_generator import PodGenerator
  File "/usr/local/lib/python3.12/site-packages/astronomer/kubernetes/pod_generator.py", line 24, in <module>
    from astronomer.kubernetes.version_compat import AIRFLOW_V_3_0_PLUS
  File "/usr/local/lib/python3.12/site-packages/astronomer/kubernetes/version_compat.py", line 41, in <module>
    from airflow.models import BaseOperator
  File "/usr/local/lib/python3.12/site-packages/airflow/models/__init__.py", line 79, in __getattr__
    val = import_string(f"{path}.{name}")
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/utils/module_loading.py", line 39, in import_string
    module = import_module(module_path)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/importlib/__init__.py", line 90, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/models/baseoperator.py", line 83, in <module>
    from airflow.models.mappedoperator import OperatorPartial, validate_mapping_kwargs
  File "/usr/local/lib/python3.12/site-packages/airflow/models/mappedoperator.py", line 54, in <module>
    from airflow.triggers.base import StartTriggerArgs
  File "/usr/local/lib/python3.12/site-packages/airflow/triggers/base.py", line 27, in <module>
    from airflow.models.taskinstance import SimpleTaskInstance
ImportError: cannot import name 'SimpleTaskInstance' from partially initialized module 'airflow.models.taskinstance' (most likely due to a circular import) (/usr/local/lib/python3.12/site-packages/airflow/models/taskinstance.py)

What you think should happen instead?

The scheduler using Kubernetes Executor should start up without issue when sentry is installed and enabled.

How to reproduce

1. Build a image with sentry-sdk installed and the following environment variables exported

export AIRFLOW__CORE__EXECUTOR="astronomer.kubernetes.executors.kubernetes_executor.KubernetesExecutor"
export AIRFLOW__SENTRY__SENTRY_ON="True"
export AIRFLOW__SENTRY__SENTRY_DSN="https://fake@fake.ingest.sentry.io/fake"
export AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="postgresql://fake:fake@fake:5432/fake"

2. Deploy and configure Kubernetes Executor

This can be reproduced with an OSS image as well.

docker run --rm apache/airflow:2.11.0 bash -c 'pip install sentry-sdk apache-airflow-providers-cncf-kubernetes==10.6.1
export AIRFLOW__CORE__EXECUTOR="KubernetesExecutor"
export AIRFLOW__SENTRY__SENTRY_ON="True"
export AIRFLOW__SENTRY__SENTRY_DSN="https://fake@fake.ingest.sentry.io/fake"
export AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="postgresql://fake:fake@fake:5432/Sentry enabled (sentry_on=True)"

python3 -c "import airflow.providers.cncf.kubernetes.executors.kubernetes_executor"

Operating System

debian

Versions of Apache Airflow Providers

No response

Deployment

Astronomer

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[wolfier]

This issue is resolved by #55809