Build DataprocCreateClusterOperator cluster_config from legacy kwargs after rendering

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

@@ -685,8 +685,13 @@ class DataprocCreateClusterOperator(GoogleCloudBaseOperator):
         "labels",
         "gcp_conn_id",
         "impersonation_chain",
+        "_legacy_cluster_kwargs",
     )
-    template_fields_renderers = {"cluster_config": "json", "virtual_cluster_config": "json"}
+    template_fields_renderers = {
+        "cluster_config": "json",
+        "virtual_cluster_config": "json",
+        "_legacy_cluster_kwargs": "json",
+    }
 
     operator_extra_links = (DataprocClusterLink(),)
 
@@ -713,6 +718,7 @@ def __init__(
         **kwargs,
     ) -> None:
         # TODO: remove one day
+        self._legacy_cluster_kwargs: dict | None = None
         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. "
@@ -732,7 +738,9 @@ def __init__(
                     "project_id argument is required when building cluster from keywords parameters"
                 )
             kwargs["project_id"] = project_id
-            cluster_config = ClusterGenerator(**kwargs).make()
+
+            # Build cluster_config in execute() after template values are rendered.
+            self._legacy_cluster_kwargs = dict(kwargs)
 
             # Remove from kwargs cluster params passed for backward compatibility
             cluster_params = inspect.signature(ClusterGenerator.__init__).parameters
@@ -897,6 +905,9 @@ def _reconcile_cluster_state(self, hook: DataprocHook, cluster: Cluster) -> Clus
         return cluster
 
     def execute(self, context: Context) -> dict:
+        if self._legacy_cluster_kwargs is not None:
+            # Build cluster_config here so it uses rendered template values.
+            self.cluster_config = ClusterGenerator(**self._legacy_cluster_kwargs).make()
 
         self.log.info("Attempting to create cluster: %s", self.cluster_name)
         hook = DataprocHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)

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

@@ -998,6 +998,25 @@ def test_deprecation_warning(self):
 
         assert op.project_id == GCP_PROJECT
         assert op.cluster_name == "cluster_name"
+        assert op.cluster_config is None
+        assert op._legacy_cluster_kwargs["num_workers"] == 2
+        assert op._legacy_cluster_kwargs["zone"] == "zone"
+
+    @mock.patch(DATAPROC_PATH.format("Cluster.to_dict"))
+    @mock.patch(DATAPROC_PATH.format("DataprocHook"))
+    def test_deprecated_kwargs_cluster_config_built_in_execute(self, mock_hook, to_dict_mock):
+        mock_hook.return_value.create_cluster.result.return_value = None
+        with pytest.warns(AirflowProviderDeprecationWarning):
+            op = DataprocCreateClusterOperator(
+                task_id=TASK_ID,
+                region=GCP_REGION,
+                project_id=GCP_PROJECT,
+                cluster_name="cluster_name",
+                num_workers=2,
+                zone="zone",
+            )
+        assert op.cluster_config is None
+        op.execute(context=self.mock_context)
         assert op.cluster_config["worker_config"]["num_instances"] == 2
         assert "zones/zone" in op.cluster_config["master_config"]["machine_type_uri"]
 
@@ -1521,6 +1540,33 @@ def test_create_execute_call_finished_before_defer(self, mock_trigger_hook, mock
         mock_hook.return_value.wait_for_operation.assert_not_called()
 
 
+@pytest.mark.db_test
+@pytest.mark.need_serialized_dag
+def test_create_cluster_operator_legacy_kwargs_survive_serialization_and_render(
+    dag_maker, create_task_instance_of_operator
+):
+    with pytest.warns(AirflowProviderDeprecationWarning):
+        ti = create_task_instance_of_operator(
+            DataprocCreateClusterOperator,
+            dag_id=TEST_DAG_ID,
+            task_id=TASK_ID,
+            region=GCP_REGION,
+            project_id=GCP_PROJECT,
+            cluster_name=CLUSTER_NAME,
+            num_workers=2,
+            zone="{{ 'templated-zone' }}",
+            gcp_conn_id=GCP_CONN_ID,
+        )
+    serialized_dag = dag_maker.get_serialized_data()
+    deserialized_dag = DagSerialization.deserialize_dag(serialized_dag["dag"])
+    deserialized_task = deserialized_dag.tasks[0]
+    assert deserialized_task._legacy_cluster_kwargs["zone"] == "{{ 'templated-zone' }}"
+
+    context = {"dag": dag_maker.dag, "ti": ti}
+    rendered_task = ti.render_templates(context=context)
+    assert rendered_task._legacy_cluster_kwargs["zone"] == "templated-zone"
+
+
 @pytest.mark.db_test
 @pytest.mark.need_serialized_dag
 def test_create_cluster_operator_extra_links(

scripts/ci/prek/validate_operators_init_exemptions.txt

@@ -12,7 +12,6 @@ providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
 providers/google/src/airflow/providers/google/cloud/operators/cloud_batch.py::CloudBatchSubmitJobOperator
 providers/google/src/airflow/providers/google/cloud/operators/cloud_build.py::CloudBuildCreateBuildOperator
 providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py::CloudDataTransferServiceCreateJobOperator
-providers/google/src/airflow/providers/google/cloud/operators/dataproc.py::DataprocCreateClusterOperator
 providers/google/src/airflow/providers/google/cloud/operators/dataproc.py::DataprocSubmitJobOperator
 providers/google/src/airflow/providers/google/cloud/operators/functions.py::CloudFunctionDeployFunctionOperator
 providers/google/src/airflow/providers/google/cloud/operators/gcs.py::GCSFileTransformOperator