feat: Add custom provider runtime checks (#39609)

Signed-off-by: Kacper Muda <mudakacper@gmail.com>

airflow/providers/openlineage/CHANGELOG.rst

@@ -26,10 +26,10 @@
 Changelog
 ---------
 
-main
+1.8.0
 .....
 
-In Airflow 2.10.0, we fix the way try_number works, so that it no longer returns different values depending on task instance state.  Importantly, after the task is done, it no longer shows current_try + 1. Thus in 1.7.2 we patch this provider to fix try_number references so they no longer adjust for the old, bad behavior.
+For Airflow >= 2.10.0, use ``apache-airflow-providers-openlineage >= 1.8.0``, the first compatible version due to changes in try_number in core Airflow (#39336). We always recommend installing the latest provider version regardless of Airflow version.
 
 1.7.1
 .....

airflow/providers/openlineage/__init__.py

@@ -29,7 +29,7 @@
 
 __all__ = ["__version__"]
 
-__version__ = "1.7.1"
+__version__ = "1.8.0"
 
 if packaging.version.parse(packaging.version.parse(airflow_version).base_version) < packaging.version.parse(
     "2.7.0"

airflow/providers/openlineage/provider.yaml

@@ -22,9 +22,10 @@ description: |
   `OpenLineage <https://openlineage.io/>`__
 
 state: ready
-source-date-epoch: 1715384461
+source-date-epoch: 1715684338
 # note that those versions are maintained by release manager - do not update them manually
 versions:
+  - 1.8.0
   - 1.7.1
   - 1.7.0
   - 1.6.0

airflow/settings.py

@@ -25,14 +25,16 @@
 import sys
 import traceback
 import warnings
+from importlib import metadata
 from typing import TYPE_CHECKING, Any, Callable
 
 import pluggy
+from packaging.version import Version
 from sqlalchemy import create_engine, exc, text
 from sqlalchemy.orm import scoped_session, sessionmaker
 from sqlalchemy.pool import NullPool
 
-from airflow import policies
+from airflow import __version__ as airflow_version, policies
 from airflow.configuration import AIRFLOW_HOME, WEBSERVER_CONFIG, conf  # noqa: F401
 from airflow.exceptions import AirflowInternalRuntimeError, RemovedInAirflow3Warning
 from airflow.executors import executor_constants
@@ -207,6 +209,31 @@ def configure_vars():
     DONOT_MODIFY_HANDLERS = conf.getboolean("logging", "donot_modify_handlers", fallback=False)
 
 
+def _run_openlineage_runtime_check():
+    """
+    Ensure compatibility of OpenLineage provider package and Airflow version.
+
+    Airflow 2.10.0 introduced some core changes (#39336) that made versions <= 1.8.0 of OpenLineage
+    provider incompatible with future Airflow versions (>= 2.10.0).
+    """
+    ol_package = "apache-airflow-providers-openlineage"
+    try:
+        ol_version = metadata.version(ol_package)
+    except metadata.PackageNotFoundError:
+        return
+
+    if ol_version and Version(ol_version) < Version("1.8.0.dev0"):
+        raise RuntimeError(
+            f"You have installed `{ol_package}` == `{ol_version}` that is not compatible with "
+            f"`apache-airflow` == `{airflow_version}`. "
+            f"For `apache-airflow` >= `2.10.0` you must use `{ol_package}` >= `1.8.0`."
+        )
+
+
+def run_providers_custom_runtime_checks():
+    _run_openlineage_runtime_check()
+
+
 class SkipDBTestsSession:
     """
     This fake session is used to skip DB tests when `_AIRFLOW_SKIP_DB_TESTS` is set.
@@ -572,6 +599,9 @@ def initialize():
     configure_orm()
     configure_action_logging()
 
+    # Run any custom runtime checks that needs to be executed for providers
+    run_providers_custom_runtime_checks()
+
     # Ensure we close DB connections at scheduler and gunicorn worker terminations
     atexit.register(dispose_orm)