Deprecate Dataproc ClusterGenerator helper class

providers/google/docs/operators/cloud/dataproc.rst

@@ -188,16 +188,12 @@ You can use deferrable mode for this action in order to run the operator asynchr
 
 Generating Cluster Config
 ^^^^^^^^^^^^^^^^^^^^^^^^^
-You can also generate **CLUSTER_CONFIG** using functional API,
-this could be easily done using **make()** of
-:class:`~airflow.providers.google.cloud.operators.dataproc.ClusterGenerator`
-You can generate and use config as followed:
 
-.. exampleinclude:: /../../google/tests/system/google/cloud/dataproc/example_dataproc_cluster_generator.py
-    :language: python
-    :dedent: 0
-    :start-after: [START how_to_cloud_dataproc_create_cluster_generate_cluster_config]
-    :end-before: [END how_to_cloud_dataproc_create_cluster_generate_cluster_config]
+.. warning::
+    **Deprecated:** The :class:`~airflow.providers.google.cloud.operators.dataproc.ClusterGenerator`
+    class is deprecated and will be removed after September 1, 2027. Please pass
+    ``cluster_config`` or ``virtual_cluster_config`` dictionaries directly as arguments to the
+    ``DataprocCreateClusterOperator``.
 
 Diagnose a cluster
 ------------------

providers/google/src/airflow/providers/google/cloud/operators/dataproc.py

@@ -22,7 +22,6 @@
 import inspect
 import re
 import time
-import warnings
 from collections.abc import MutableSequence, Sequence
 from dataclasses import dataclass
 from datetime import datetime, timedelta
@@ -62,6 +61,7 @@
     DataprocSubmitTrigger,
 )
 from airflow.providers.google.cloud.utils.dataproc import DataprocOperationType
+from airflow.providers.google.common.deprecated import deprecated
 from airflow.providers.google.common.hooks.base_google import PROVIDE_PROJECT_ID
 from airflow.triggers.base import StartTriggerArgs
 
@@ -117,6 +117,14 @@ class InstanceFlexibilityPolicy:
     instance_selection_list: list[InstanceSelection]
 
 
+@deprecated(
+    planned_removal_date="September 1, 2027",
+    reason="Since passing cluster parameters by keyword in the DataprocCreateClusterOperator is scheduled "
+    "for deletion, there is no longer a need to maintain the 'ClusterGenerator' helper class.",
+    instructions="Please pass 'cluster_config' or 'virtual_cluster_config' directly as "
+    "DataprocCreateClusterOperator arguments.",
+    category=AirflowProviderDeprecationWarning,
+)
 class ClusterGenerator:
     """
     Create a new Dataproc Cluster.
@@ -712,16 +720,7 @@ def __init__(
         polling_interval_seconds: int = 10,
         **kwargs,
     ) -> None:
-        # TODO: remove one day
         if cluster_config is None and virtual_cluster_config is None:
-            warnings.warn(
-                f"Passing cluster parameters by keywords to `{type(self).__name__}` will be deprecated. "
-                "Please provide cluster_config object using `cluster_config` parameter. "
-                "You can use `airflow.dataproc.ClusterGenerator.generate_cluster` "
-                "method to obtain cluster object. Planned removal date: October 5, 2026.",
-                AirflowProviderDeprecationWarning,
-                stacklevel=2,
-            )
             # Remove result of apply defaults
             if "params" in kwargs:
                 del kwargs["params"]

providers/google/tests/system/google/cloud/dataproc/example_dataproc_batch_persistent.py

@@ -27,7 +27,6 @@
 
 from airflow.models.dag import DAG
 from airflow.providers.google.cloud.operators.dataproc import (
-    ClusterGenerator,
     DataprocCreateBatchOperator,
     DataprocCreateClusterOperator,
     DataprocDeleteBatchOperator,
@@ -53,17 +52,26 @@
 CLUSTER_NAME = CLUSTER_NAME_BASE if len(CLUSTER_NAME_FULL) >= 33 else CLUSTER_NAME_FULL
 BATCH_ID = f"batch-{ENV_ID}-{DAG_ID}".replace("_", "-")
 
-CLUSTER_GENERATOR_CONFIG_FOR_PHS = ClusterGenerator(
-    project_id=PROJECT_ID,
-    region=REGION,
-    master_machine_type="n1-standard-4",
-    worker_machine_type="n1-standard-4",
-    num_workers=0,
-    properties={
-        "spark:spark.history.fs.logDirectory": f"gs://{BUCKET_NAME}",
+CLUSTER_CONFIG_FOR_PHS = {
+    "master_config": {
+        "num_instances": 1,
+        "machine_type_uri": "n1-standard-4",
     },
-    enable_component_gateway=True,
-).make()
+    "worker_config": {
+        "num_instances": 0,
+        "machine_type_uri": "n1-standard-4",
+    },
+    "software_config": {
+        "properties": {
+            "dataproc:dataproc.allow.zero.workers": "true",
+            "spark:spark.history.fs.logDirectory": f"gs://{BUCKET_NAME}",
+        },
+    },
+    "endpoint_config": {
+        "enable_http_port_access": True,
+    },
+}
+
 BATCH_CONFIG_WITH_PHS = {
     "spark_batch": {
         "jar_file_uris": ["file:///usr/lib/spark/examples/jars/spark-examples.jar"],
@@ -94,7 +102,7 @@
     create_cluster = DataprocCreateClusterOperator(
         task_id="create_cluster_for_phs",
         project_id=PROJECT_ID,
-        cluster_config=CLUSTER_GENERATOR_CONFIG_FOR_PHS,
+        cluster_config=CLUSTER_CONFIG_FOR_PHS,
         region=REGION,
         cluster_name=CLUSTER_NAME,
         retry=Retry(maximum=100.0, initial=10.0, multiplier=1.0),

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

@@ -632,6 +632,7 @@ def setup_class(cls):
             ]
 
 
+@pytest.mark.filterwarnings("ignore::airflow.exceptions.AirflowProviderDeprecationWarning")
 class TestsClusterGenerator:
     def test_image_version(self):
         with pytest.raises(ValueError, match="custom_image and image_version"):
@@ -982,6 +983,10 @@ def test_build_with_engine(self):
         cluster = generator.make()
         assert cluster["engine"] == "DEFAULT"
 
+    def test_deprecation_warning(self):
+        with pytest.warns(AirflowProviderDeprecationWarning, match="ClusterGenerator"):
+            ClusterGenerator(project_id=GCP_PROJECT)
+
 
 class TestDataprocCreateClusterOperator(DataprocClusterTestBase):
     def test_deprecation_warning(self):
@@ -994,7 +999,7 @@ def test_deprecation_warning(self):
                 num_workers=2,
                 zone="zone",
             )
-        assert_warning("Passing cluster parameters by keywords", warnings)
+        assert_warning("Since passing cluster parameters by keyword", warnings)
 
         assert op.project_id == GCP_PROJECT
         assert op.cluster_name == "cluster_name"