Revert "Suspend Apache Beam Provider due to grpcio limitation (#61926)"

This reverts commit 917abea.

- Remove hacks regarding the beam provider suspension, as they are not
  relevant anymore.

ISSUE: #66551

Author: Oleg Kachur

airflow-core/tests/unit/always/test_example_dags.py

@@ -69,17 +69,6 @@
     # Ray uses pydantic v1 internally, which fails to infer types in Python 3.14.
     # TODO: remove once ray releases a version with Python 3.14 support.
     "providers/google/tests/system/google/cloud/ray/example_ray_job.py",
-    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_go.py",
-    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_java_streaming.py",
-    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_native_java.py",
-    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_native_python.py",
-    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_native_python_async.py",
-    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_pipeline.py",
-    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_pipeline_streaming.py",
-    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_sensors_deferrable.py",
-    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_streaming_python.py",
-    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_template.py",
-    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_yaml.py",
     "providers/google/tests/system/google/cloud/gcs/example_firestore.py",
 )
 

dev/breeze/tests/test_packages.py

@@ -118,12 +118,12 @@ def test_get_removed_providers():
 
 def test_get_suspended_provider_ids():
     # Modify it every time we suspend/resume provider
-    assert get_suspended_provider_ids() == ["apache.beam"]
+    assert get_suspended_provider_ids() == []
 
 
 def test_get_suspended_provider_folders():
     # Modify it every time we suspend/resume provider
-    assert get_suspended_provider_folders() == ["apache/beam"]
+    assert get_suspended_provider_folders() == []
 
 
 @pytest.mark.parametrize(

devel-common/src/tests_common/test_utils/providers.py

@@ -68,7 +68,7 @@ def get_provider_min_airflow_version(provider_name: str) -> tuple[int, ...]:
 
 
 # Ignore module import errors for any suspended provider paths used in example dags
-IGNORE_MODULE_IMPORT_ERRORS: list[str] = ["airflow.providers.apache.beam"]
+IGNORE_MODULE_IMPORT_ERRORS: list[str] = []
 
 
 def get_suspended_providers_folders() -> list[str]:

providers/apache/beam/provider.yaml

@@ -21,9 +21,9 @@ name: Apache Beam
 description: |
     `Apache Beam <https://beam.apache.org/>`__.
 
-state: suspended
-source-date-epoch: 1768334134
+state: ready
 lifecycle: production
+source-date-epoch: 1772063928
 # Note that those versions are maintained by release manager - do not update them manually
 # with the exception of case where other provider in sources has >= new provider version.
 # In such case adding >= NEW_VERSION and bumping to NEW_VERSION in a provider have

providers/google/pyproject.toml

@@ -149,6 +149,9 @@ dependencies = [
 # The optional dependencies should be modified in place in the generated file
 # Any change in the dependencies is preserved when the file is regenerated
 [project.optional-dependencies]
+"apache.beam" = [
+    "apache-airflow-providers-apache-beam>=6.2.2",
+]
 "cncf.kubernetes" = [
     "apache-airflow-providers-cncf-kubernetes>=10.1.0",
 ]
@@ -216,6 +219,7 @@ dev = [
     "apache-airflow-task-sdk",
     "apache-airflow-devel-common",
     "apache-airflow-providers-amazon",
+    "apache-airflow-providers-apache-beam",
     "apache-airflow-providers-apache-cassandra",
     "apache-airflow-providers-cncf-kubernetes",
     "apache-airflow-providers-common-compat",

providers/google/tests/unit/google/cloud/hooks/test_dataflow.py

@@ -28,11 +28,6 @@
 from uuid import UUID
 
 import pytest
-
-# TODO: Remove below skip once beam provider changed to ready state
-pytest.importorskip("apache-beam", reason="apache-beam package suspended due to grpcio limitation")
-
-
 from google.cloud.dataflow_v1beta3 import (
     GetJobMetricsRequest,
     GetJobRequest,

providers/google/tests/unit/google/cloud/operators/test_dataflow.py

@@ -21,10 +21,6 @@
 
 import httplib2
 import pytest
-
-# TODO: Remove below skip once beam provider changed to ready state
-pytest.importorskip("apache-beam", reason="apache-beam package suspended due to grpcio limitation")
-
 from googleapiclient.errors import HttpError
 
 from airflow.providers.common.compat.sdk import AirflowException

providers/google/tests/unit/google/cloud/operators/test_datapipeline.py

@@ -21,9 +21,6 @@
 
 import pytest
 
-# TODO: Remove below skip once beam provider changed to ready state
-pytest.importorskip("apache-beam", reason="apache-beam package suspended due to grpcio limitation")
-
 from airflow.providers.google.cloud.operators.dataflow import (
     DataflowCreatePipelineOperator,
     DataflowRunPipelineOperator,

providers/google/tests/unit/google/cloud/sensors/test_dataflow.py

@@ -21,9 +21,6 @@
 
 import pytest
 
-# TODO: Remove below skip once beam provider changed to ready state
-pytest.importorskip("apache-beam", reason="apache-beam package suspended due to grpcio limitation")
-
 from airflow.providers.common.compat.sdk import AirflowException, TaskDeferred
 from airflow.providers.google.cloud.hooks.dataflow import DataflowJobStatus
 from airflow.providers.google.cloud.sensors.dataflow import (

providers/google/tests/unit/google/cloud/triggers/test_dataflow.py

@@ -19,58 +19,14 @@
 
 import asyncio
 import logging
-import sys
-import types
 from unittest import mock
 
 import pytest
 from google.api_core.exceptions import ServiceUnavailable
 from google.cloud.dataflow_v1beta3 import Job, JobState, JobType
 
-
-# While the apache-beam provider is suspended (#61926),
-# `airflow.providers.google.cloud.hooks.dataflow` cannot be imported because it
-# does `from airflow.providers.apache.beam.hooks.beam import ...` at module
-# scope. Probe-import the real module first (which also triggers Airflow's
-# provider-entry-point discovery while the real `airflow.providers.apache`
-# namespace package is still intact) and only fall back to stubs when the
-# probe fails. The stubs deliberately cover only the missing beam-specific
-# subtree — we never overwrite `airflow.providers.apache` itself, otherwise
-# sibling providers such as `airflow.providers.apache.pig` would be unreachable
-# via the namespace package's `__path__`.
-def _beam_module_stubs() -> dict[str, types.ModuleType]:
-    beam_package = types.ModuleType("airflow.providers.apache.beam")
-    beam_package.__path__ = []
-    hooks_package = types.ModuleType("airflow.providers.apache.beam.hooks")
-    hooks_package.__path__ = []
-
-    class BeamRunnerType:
-        DataflowRunner = "DataflowRunner"
-
-    class BeamModule(types.ModuleType):
-        BeamHook: object
-        BeamRunnerType: object
-        beam_options_to_args: object
-
-    beam_module = BeamModule("airflow.providers.apache.beam.hooks.beam")
-    beam_module.BeamHook = mock.MagicMock()
-    beam_module.BeamRunnerType = BeamRunnerType
-    beam_module.beam_options_to_args = mock.MagicMock(return_value=[])
-
-    return {
-        "airflow.providers.apache.beam": beam_package,
-        "airflow.providers.apache.beam.hooks": hooks_package,
-        "airflow.providers.apache.beam.hooks.beam": beam_module,
-    }
-
-
-try:
-    import airflow.providers.apache.beam.hooks.beam  # noqa: F401
-except ImportError:
-    sys.modules.update(_beam_module_stubs())
-
-from airflow.providers.google.cloud.hooks.dataflow import DataflowJobStatus  # noqa: E402
-from airflow.providers.google.cloud.triggers.dataflow import (  # noqa: E402
+from airflow.providers.google.cloud.hooks.dataflow import DataflowJobStatus
+from airflow.providers.google.cloud.triggers.dataflow import (
     DataflowJobAutoScalingEventTrigger,
     DataflowJobMessagesTrigger,
     DataflowJobMetricsTrigger,
@@ -79,7 +35,7 @@ class BeamModule(types.ModuleType):
     DataflowStartYamlJobTrigger,
     TemplateJobStartTrigger,
 )
-from airflow.triggers.base import TriggerEvent  # noqa: E402
+from airflow.triggers.base import TriggerEvent
 
 PROJECT_ID = "test-project-id"
 JOB_ID = "test_job_id_2012-12-23-10:00"