Do not ignore the internal_ip_only if set to false in Dataproc cluste…

…r config (#37014)

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

@@ -361,10 +361,10 @@ def _build_gce_cluster_config(self, cluster_data):
         if self.subnetwork_uri:
             cluster_data[config]["subnetwork_uri"] = self.subnetwork_uri
 
-        if self.internal_ip_only:
-            if not self.subnetwork_uri:
+        if self.internal_ip_only is not None:
+            if not self.subnetwork_uri and self.internal_ip_only:
                 raise AirflowException("Set internal_ip_only to true only when you pass a subnetwork_uri.")
-            cluster_data[config]["internal_ip_only"] = True
+            cluster_data[config]["internal_ip_only"] = self.internal_ip_only
 
         if self.tags:
             cluster_data[config]["tags"] = self.tags

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

@@ -677,6 +677,19 @@ def test_build_with_gpu_accelerator(self):
         cluster = generator.make()
         assert CONFIG_WITH_GPU_ACCELERATOR == cluster
 
+    def test_build_with_default_value_for_internal_ip_only(self):
+        generator = ClusterGenerator(project_id="project_id")
+        cluster = generator.make()
+        assert "internal_ip_only" not in cluster["gce_cluster_config"]
+
+    def test_build_sets_provided_value_for_internal_ip_only(self):
+        for internal_ip_only in [True, False]:
+            generator = ClusterGenerator(
+                project_id="project_id", internal_ip_only=internal_ip_only, subnetwork_uri="subnetwork_uri"
+            )
+            cluster = generator.make()
+            assert cluster["gce_cluster_config"]["internal_ip_only"] == internal_ip_only
+
 
 class TestDataprocCreateClusterOperator(DataprocClusterTestBase):
     def test_deprecation_warning(self):