Dataproc: Remove default value of region (#23350)

* `region` parameter has no default value.
  affected functions/classes:
  `DataprocHook.cancel_job`
  `DataprocCreateClusterOperator`
  `DataprocJobBaseOperator`

  * `DataprocJobBaseOperator`: order of parameters has changed

Author: eladkal

airflow/providers/google/CHANGELOG.rst

@@ -34,6 +34,14 @@ Breaking changes
    For more information, see `Deprecation and sunset <https://developers.google.com/google-ads/api/docs/sunset-dates>`_
    and `Upgrading to the newest version <https://developers.google.com/google-ads/api/docs/version-migration>`_
 
+* ``DataprocJobBaseOperator``: order of parameters has changed
+
+* ``region`` parameter has no default value
+  affected functions/classes:
+  ``DataprocHook.cancel_job``
+  ``DataprocCreateClusterOperator``
+  ``DataprocJobBaseOperator``
+
 * ``DatastoreHook``: Remove ``datastore_conn_id``. Please use ``gcp_conn_id``
 
 * ``CloudBuildCreateBuildOperator``: Remove ``body``. Please use ``build``

airflow/providers/google/cloud/hooks/dataproc.py

@@ -810,13 +810,6 @@ def cancel_job(
             ``retry`` is specified, the timeout applies to each individual attempt.
         :param metadata: Additional metadata that is provided to the method.
         """
-        if region is None:
-            warnings.warn(
-                "Default region value `global` will be deprecated. Please, provide region value.",
-                DeprecationWarning,
-                stacklevel=2,
-            )
-            region = 'global'
         client = self.get_job_client(region=region)
 
         job = client.cancel_job(

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

@@ -456,7 +456,7 @@ def __init__(
         self,
         *,
         cluster_name: str,
-        region: Optional[str] = None,
+        region: str,
         project_id: Optional[str] = None,
         cluster_config: Optional[Union[Dict, Cluster]] = None,
         virtual_cluster_config: Optional[Dict] = None,
@@ -471,13 +471,6 @@ def __init__(
         impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
         **kwargs,
     ) -> None:
-        if region is None:
-            warnings.warn(
-                "Default region value `global` will be deprecated. Please, provide region value.",
-                DeprecationWarning,
-                stacklevel=2,
-            )
-            region = 'global'
 
         # TODO: remove one day
         if cluster_config is None and virtual_cluster_config is None:
@@ -839,6 +832,7 @@ class DataprocJobBaseOperator(BaseOperator):
     """
     The base class for operators that launch job on DataProc.
 
+    :param region: The specified region where the dataproc cluster is created.
     :param job_name: The job name used in the DataProc cluster. This name by default
         is the task_id appended with the execution data, but can be templated. The
         name will always be appended with a random number to avoid name clashes.
@@ -856,7 +850,6 @@ class DataprocJobBaseOperator(BaseOperator):
     :param labels: The labels to associate with this job. Label keys must contain 1 to 63 characters,
         and must conform to RFC 1035. Label values may be empty, but, if present, must contain 1 to 63
         characters, and must conform to RFC 1035. No more than 32 labels can be associated with a job.
-    :param region: The specified region where the dataproc cluster is created.
     :param job_error_states: Job states that should be considered error states.
         Any states in this set will result in an error being raised and failure of the
         task. Eg, if the ``CANCELLED`` state should also be considered a task failure,
@@ -889,6 +882,7 @@ class DataprocJobBaseOperator(BaseOperator):
     def __init__(
         self,
         *,
+        region: str,
         job_name: str = '{{task.task_id}}_{{ds_nodash}}',
         cluster_name: str = "cluster-1",
         project_id: Optional[str] = None,
@@ -897,7 +891,6 @@ def __init__(
         gcp_conn_id: str = 'google_cloud_default',
         delegate_to: Optional[str] = None,
         labels: Optional[Dict] = None,
-        region: Optional[str] = None,
         job_error_states: Optional[Set[str]] = None,
         impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
         asynchronous: bool = False,
@@ -911,14 +904,6 @@ def __init__(
         self.cluster_name = cluster_name
         self.dataproc_properties = dataproc_properties
         self.dataproc_jars = dataproc_jars
-
-        if region is None:
-            warnings.warn(
-                "Default region value `global` will be deprecated. Please, provide region value.",
-                DeprecationWarning,
-                stacklevel=2,
-            )
-            region = 'global'
         self.region = region
 
         self.job_error_states = job_error_states if job_error_states is not None else {'ERROR'}

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

@@ -414,16 +414,6 @@ def test_deprecation_warning(self):
         assert op.cluster_config['worker_config']['num_instances'] == 2
         assert "zones/zone" in op.cluster_config['master_config']["machine_type_uri"]
 
-        with pytest.warns(DeprecationWarning) as warnings:
-            op_default_region = DataprocCreateClusterOperator(
-                task_id=TASK_ID,
-                project_id=GCP_PROJECT,
-                cluster_name="cluster_name",
-                cluster_config=op.cluster_config,
-            )
-        assert_warning("Default region value", warnings)
-        assert op_default_region.region == 'global'
-
     @mock.patch(DATAPROC_PATH.format("Cluster.to_dict"))
     @mock.patch(DATAPROC_PATH.format("DataprocHook"))
     def test_execute(self, mock_hook, to_dict_mock):