diff --git a/google/cloud/dataproc_v1/services/autoscaling_policy_service/async_client.py b/google/cloud/dataproc_v1/services/autoscaling_policy_service/async_client.py
index f2c5718c..76e7ff5c 100644
--- a/google/cloud/dataproc_v1/services/autoscaling_policy_service/async_client.py
+++ b/google/cloud/dataproc_v1/services/autoscaling_policy_service/async_client.py
@@ -621,6 +621,12 @@ async def delete_autoscaling_policy(
             request, retry=retry, timeout=timeout, metadata=metadata,
         )
 
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.transport.close()
+
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/google/cloud/dataproc_v1/services/autoscaling_policy_service/client.py b/google/cloud/dataproc_v1/services/autoscaling_policy_service/client.py
index 4d3e5aad..43a042b8 100644
--- a/google/cloud/dataproc_v1/services/autoscaling_policy_service/client.py
+++ b/google/cloud/dataproc_v1/services/autoscaling_policy_service/client.py
@@ -350,10 +350,7 @@ def __init__(
                 client_cert_source_for_mtls=client_cert_source_func,
                 quota_project_id=client_options.quota_project_id,
                 client_info=client_info,
-                always_use_jwt_access=(
-                    Transport == type(self).get_transport_class("grpc")
-                    or Transport == type(self).get_transport_class("grpc_asyncio")
-                ),
+                always_use_jwt_access=True,
             )
 
     def create_autoscaling_policy(
@@ -791,6 +788,19 @@ def delete_autoscaling_policy(
             request, retry=retry, timeout=timeout, metadata=metadata,
         )
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        """Releases underlying transport's resources.
+
+        .. warning::
+            ONLY use as a context manager if the transport is NOT shared
+            with other clients! Exiting the with block will CLOSE the transport
+            and may cause errors in other clients!
+        """
+        self.transport.close()
+
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/google/cloud/dataproc_v1/services/autoscaling_policy_service/transports/base.py b/google/cloud/dataproc_v1/services/autoscaling_policy_service/transports/base.py
index 9c988099..bcaddf6c 100644
--- a/google/cloud/dataproc_v1/services/autoscaling_policy_service/transports/base.py
+++ b/google/cloud/dataproc_v1/services/autoscaling_policy_service/transports/base.py
@@ -210,6 +210,15 @@ def _prep_wrapped_messages(self, client_info):
             ),
         }
 
+    def close(self):
+        """Closes resources associated with the transport.
+
+       .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
+
     @property
     def create_autoscaling_policy(
         self,
diff --git a/google/cloud/dataproc_v1/services/autoscaling_policy_service/transports/grpc.py b/google/cloud/dataproc_v1/services/autoscaling_policy_service/transports/grpc.py
index 83931784..4d5eefcd 100644
--- a/google/cloud/dataproc_v1/services/autoscaling_policy_service/transports/grpc.py
+++ b/google/cloud/dataproc_v1/services/autoscaling_policy_service/transports/grpc.py
@@ -376,5 +376,8 @@ def delete_autoscaling_policy(
             )
         return self._stubs["delete_autoscaling_policy"]
 
+    def close(self):
+        self.grpc_channel.close()
+
 
 __all__ = ("AutoscalingPolicyServiceGrpcTransport",)
diff --git a/google/cloud/dataproc_v1/services/autoscaling_policy_service/transports/grpc_asyncio.py b/google/cloud/dataproc_v1/services/autoscaling_policy_service/transports/grpc_asyncio.py
index ae024621..07c82123 100644
--- a/google/cloud/dataproc_v1/services/autoscaling_policy_service/transports/grpc_asyncio.py
+++ b/google/cloud/dataproc_v1/services/autoscaling_policy_service/transports/grpc_asyncio.py
@@ -380,5 +380,8 @@ def delete_autoscaling_policy(
             )
         return self._stubs["delete_autoscaling_policy"]
 
+    def close(self):
+        return self.grpc_channel.close()
+
 
 __all__ = ("AutoscalingPolicyServiceGrpcAsyncIOTransport",)
diff --git a/google/cloud/dataproc_v1/services/cluster_controller/async_client.py b/google/cloud/dataproc_v1/services/cluster_controller/async_client.py
index 67dc8a57..57ca5e8a 100644
--- a/google/cloud/dataproc_v1/services/cluster_controller/async_client.py
+++ b/google/cloud/dataproc_v1/services/cluster_controller/async_client.py
@@ -1015,6 +1015,12 @@ async def diagnose_cluster(
         # Done; return the response.
         return response
 
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.transport.close()
+
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/google/cloud/dataproc_v1/services/cluster_controller/client.py b/google/cloud/dataproc_v1/services/cluster_controller/client.py
index 870eb9e9..f6659382 100644
--- a/google/cloud/dataproc_v1/services/cluster_controller/client.py
+++ b/google/cloud/dataproc_v1/services/cluster_controller/client.py
@@ -369,10 +369,7 @@ def __init__(
                 client_cert_source_for_mtls=client_cert_source_func,
                 quota_project_id=client_options.quota_project_id,
                 client_info=client_info,
-                always_use_jwt_access=(
-                    Transport == type(self).get_transport_class("grpc")
-                    or Transport == type(self).get_transport_class("grpc_asyncio")
-                ),
+                always_use_jwt_access=True,
             )
 
     def create_cluster(
@@ -1164,6 +1161,19 @@ def diagnose_cluster(
         # Done; return the response.
         return response
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        """Releases underlying transport's resources.
+
+        .. warning::
+            ONLY use as a context manager if the transport is NOT shared
+            with other clients! Exiting the with block will CLOSE the transport
+            and may cause errors in other clients!
+        """
+        self.transport.close()
+
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/google/cloud/dataproc_v1/services/cluster_controller/transports/base.py b/google/cloud/dataproc_v1/services/cluster_controller/transports/base.py
index 508158ed..880dc1db 100644
--- a/google/cloud/dataproc_v1/services/cluster_controller/transports/base.py
+++ b/google/cloud/dataproc_v1/services/cluster_controller/transports/base.py
@@ -250,6 +250,15 @@ def _prep_wrapped_messages(self, client_info):
             ),
         }
 
+    def close(self):
+        """Closes resources associated with the transport.
+
+       .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
+
     @property
     def operations_client(self) -> operations_v1.OperationsClient:
         """Return the client designed to process long-running operations."""
diff --git a/google/cloud/dataproc_v1/services/cluster_controller/transports/grpc.py b/google/cloud/dataproc_v1/services/cluster_controller/transports/grpc.py
index 67d97c46..19283747 100644
--- a/google/cloud/dataproc_v1/services/cluster_controller/transports/grpc.py
+++ b/google/cloud/dataproc_v1/services/cluster_controller/transports/grpc.py
@@ -467,5 +467,8 @@ def diagnose_cluster(
             )
         return self._stubs["diagnose_cluster"]
 
+    def close(self):
+        self.grpc_channel.close()
+
 
 __all__ = ("ClusterControllerGrpcTransport",)
diff --git a/google/cloud/dataproc_v1/services/cluster_controller/transports/grpc_asyncio.py b/google/cloud/dataproc_v1/services/cluster_controller/transports/grpc_asyncio.py
index e19d261b..b3def8d7 100644
--- a/google/cloud/dataproc_v1/services/cluster_controller/transports/grpc_asyncio.py
+++ b/google/cloud/dataproc_v1/services/cluster_controller/transports/grpc_asyncio.py
@@ -478,5 +478,8 @@ def diagnose_cluster(
             )
         return self._stubs["diagnose_cluster"]
 
+    def close(self):
+        return self.grpc_channel.close()
+
 
 __all__ = ("ClusterControllerGrpcAsyncIOTransport",)
diff --git a/google/cloud/dataproc_v1/services/job_controller/async_client.py b/google/cloud/dataproc_v1/services/job_controller/async_client.py
index 809b1178..cb82f02e 100644
--- a/google/cloud/dataproc_v1/services/job_controller/async_client.py
+++ b/google/cloud/dataproc_v1/services/job_controller/async_client.py
@@ -798,6 +798,12 @@ async def delete_job(
             request, retry=retry, timeout=timeout, metadata=metadata,
         )
 
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.transport.close()
+
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/google/cloud/dataproc_v1/services/job_controller/client.py b/google/cloud/dataproc_v1/services/job_controller/client.py
index 1648b69c..e02935e6 100644
--- a/google/cloud/dataproc_v1/services/job_controller/client.py
+++ b/google/cloud/dataproc_v1/services/job_controller/client.py
@@ -328,10 +328,7 @@ def __init__(
                 client_cert_source_for_mtls=client_cert_source_func,
                 quota_project_id=client_options.quota_project_id,
                 client_info=client_info,
-                always_use_jwt_access=(
-                    Transport == type(self).get_transport_class("grpc")
-                    or Transport == type(self).get_transport_class("grpc_asyncio")
-                ),
+                always_use_jwt_access=True,
             )
 
     def submit_job(
@@ -903,6 +900,19 @@ def delete_job(
             request, retry=retry, timeout=timeout, metadata=metadata,
         )
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        """Releases underlying transport's resources.
+
+        .. warning::
+            ONLY use as a context manager if the transport is NOT shared
+            with other clients! Exiting the with block will CLOSE the transport
+            and may cause errors in other clients!
+        """
+        self.transport.close()
+
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/google/cloud/dataproc_v1/services/job_controller/transports/base.py b/google/cloud/dataproc_v1/services/job_controller/transports/base.py
index de2683cd..60fe2244 100644
--- a/google/cloud/dataproc_v1/services/job_controller/transports/base.py
+++ b/google/cloud/dataproc_v1/services/job_controller/transports/base.py
@@ -261,6 +261,15 @@ def _prep_wrapped_messages(self, client_info):
             ),
         }
 
+    def close(self):
+        """Closes resources associated with the transport.
+
+       .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
+
     @property
     def operations_client(self) -> operations_v1.OperationsClient:
         """Return the client designed to process long-running operations."""
diff --git a/google/cloud/dataproc_v1/services/job_controller/transports/grpc.py b/google/cloud/dataproc_v1/services/job_controller/transports/grpc.py
index e20d4103..b69f64b4 100644
--- a/google/cloud/dataproc_v1/services/job_controller/transports/grpc.py
+++ b/google/cloud/dataproc_v1/services/job_controller/transports/grpc.py
@@ -419,5 +419,8 @@ def delete_job(self) -> Callable[[jobs.DeleteJobRequest], empty_pb2.Empty]:
             )
         return self._stubs["delete_job"]
 
+    def close(self):
+        self.grpc_channel.close()
+
 
 __all__ = ("JobControllerGrpcTransport",)
diff --git a/google/cloud/dataproc_v1/services/job_controller/transports/grpc_asyncio.py b/google/cloud/dataproc_v1/services/job_controller/transports/grpc_asyncio.py
index 59de6e20..4c9f9d3a 100644
--- a/google/cloud/dataproc_v1/services/job_controller/transports/grpc_asyncio.py
+++ b/google/cloud/dataproc_v1/services/job_controller/transports/grpc_asyncio.py
@@ -428,5 +428,8 @@ def delete_job(
             )
         return self._stubs["delete_job"]
 
+    def close(self):
+        return self.grpc_channel.close()
+
 
 __all__ = ("JobControllerGrpcAsyncIOTransport",)
diff --git a/google/cloud/dataproc_v1/services/workflow_template_service/async_client.py b/google/cloud/dataproc_v1/services/workflow_template_service/async_client.py
index 6e895241..294b27eb 100644
--- a/google/cloud/dataproc_v1/services/workflow_template_service/async_client.py
+++ b/google/cloud/dataproc_v1/services/workflow_template_service/async_client.py
@@ -948,6 +948,12 @@ async def delete_workflow_template(
             request, retry=retry, timeout=timeout, metadata=metadata,
         )
 
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.transport.close()
+
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/google/cloud/dataproc_v1/services/workflow_template_service/client.py b/google/cloud/dataproc_v1/services/workflow_template_service/client.py
index 7725b1ad..c11b5499 100644
--- a/google/cloud/dataproc_v1/services/workflow_template_service/client.py
+++ b/google/cloud/dataproc_v1/services/workflow_template_service/client.py
@@ -387,10 +387,7 @@ def __init__(
                 client_cert_source_for_mtls=client_cert_source_func,
                 quota_project_id=client_options.quota_project_id,
                 client_info=client_info,
-                always_use_jwt_access=(
-                    Transport == type(self).get_transport_class("grpc")
-                    or Transport == type(self).get_transport_class("grpc_asyncio")
-                ),
+                always_use_jwt_access=True,
             )
 
     def create_workflow_template(
@@ -1106,6 +1103,19 @@ def delete_workflow_template(
             request, retry=retry, timeout=timeout, metadata=metadata,
         )
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        """Releases underlying transport's resources.
+
+        .. warning::
+            ONLY use as a context manager if the transport is NOT shared
+            with other clients! Exiting the with block will CLOSE the transport
+            and may cause errors in other clients!
+        """
+        self.transport.close()
+
 
 try:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
diff --git a/google/cloud/dataproc_v1/services/workflow_template_service/transports/base.py b/google/cloud/dataproc_v1/services/workflow_template_service/transports/base.py
index 6991402f..343b96c2 100644
--- a/google/cloud/dataproc_v1/services/workflow_template_service/transports/base.py
+++ b/google/cloud/dataproc_v1/services/workflow_template_service/transports/base.py
@@ -259,6 +259,15 @@ def _prep_wrapped_messages(self, client_info):
             ),
         }
 
+    def close(self):
+        """Closes resources associated with the transport.
+
+       .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
+
     @property
     def operations_client(self) -> operations_v1.OperationsClient:
         """Return the client designed to process long-running operations."""
diff --git a/google/cloud/dataproc_v1/services/workflow_template_service/transports/grpc.py b/google/cloud/dataproc_v1/services/workflow_template_service/transports/grpc.py
index d609525c..fd20ffed 100644
--- a/google/cloud/dataproc_v1/services/workflow_template_service/transports/grpc.py
+++ b/google/cloud/dataproc_v1/services/workflow_template_service/transports/grpc.py
@@ -500,5 +500,8 @@ def delete_workflow_template(
             )
         return self._stubs["delete_workflow_template"]
 
+    def close(self):
+        self.grpc_channel.close()
+
 
 __all__ = ("WorkflowTemplateServiceGrpcTransport",)
diff --git a/google/cloud/dataproc_v1/services/workflow_template_service/transports/grpc_asyncio.py b/google/cloud/dataproc_v1/services/workflow_template_service/transports/grpc_asyncio.py
index c9a93367..729d8f41 100644
--- a/google/cloud/dataproc_v1/services/workflow_template_service/transports/grpc_asyncio.py
+++ b/google/cloud/dataproc_v1/services/workflow_template_service/transports/grpc_asyncio.py
@@ -507,5 +507,8 @@ def delete_workflow_template(
             )
         return self._stubs["delete_workflow_template"]
 
+    def close(self):
+        return self.grpc_channel.close()
+
 
 __all__ = ("WorkflowTemplateServiceGrpcAsyncIOTransport",)
diff --git a/google/cloud/dataproc_v1/types/autoscaling_policies.py b/google/cloud/dataproc_v1/types/autoscaling_policies.py
index ae9afa52..2b10f9cc 100644
--- a/google/cloud/dataproc_v1/types/autoscaling_policies.py
+++ b/google/cloud/dataproc_v1/types/autoscaling_policies.py
@@ -84,6 +84,7 @@ class AutoscalingPolicy(proto.Message):
 
 class BasicAutoscalingAlgorithm(proto.Message):
     r"""Basic algorithm for autoscaling.
+
     Attributes:
         yarn_config (google.cloud.dataproc_v1.types.BasicYarnAutoscalingConfig):
             Required. YARN autoscaling configuration.
@@ -105,6 +106,7 @@ class BasicAutoscalingAlgorithm(proto.Message):
 
 class BasicYarnAutoscalingConfig(proto.Message):
     r"""Basic autoscaling configurations for YARN.
+
     Attributes:
         graceful_decommission_timeout (google.protobuf.duration_pb2.Duration):
             Required. Timeout for YARN graceful decommissioning of Node
@@ -216,6 +218,7 @@ class InstanceGroupAutoscalingPolicyConfig(proto.Message):
 
 class CreateAutoscalingPolicyRequest(proto.Message):
     r"""A request to create an autoscaling policy.
+
     Attributes:
         parent (str):
             Required. The "resource name" of the region or location, as
@@ -239,6 +242,7 @@ class CreateAutoscalingPolicyRequest(proto.Message):
 
 class GetAutoscalingPolicyRequest(proto.Message):
     r"""A request to fetch an autoscaling policy.
+
     Attributes:
         name (str):
             Required. The "resource name" of the autoscaling policy, as
@@ -259,6 +263,7 @@ class GetAutoscalingPolicyRequest(proto.Message):
 
 class UpdateAutoscalingPolicyRequest(proto.Message):
     r"""A request to update an autoscaling policy.
+
     Attributes:
         policy (google.cloud.dataproc_v1.types.AutoscalingPolicy):
             Required. The updated autoscaling policy.
@@ -292,6 +297,7 @@ class DeleteAutoscalingPolicyRequest(proto.Message):
 
 class ListAutoscalingPoliciesRequest(proto.Message):
     r"""A request to list autoscaling policies in a project.
+
     Attributes:
         parent (str):
             Required. The "resource name" of the region or location, as
diff --git a/google/cloud/dataproc_v1/types/clusters.py b/google/cloud/dataproc_v1/types/clusters.py
index f2837680..4e52ca9f 100644
--- a/google/cloud/dataproc_v1/types/clusters.py
+++ b/google/cloud/dataproc_v1/types/clusters.py
@@ -115,6 +115,7 @@ class Cluster(proto.Message):
 
 class ClusterConfig(proto.Message):
     r"""The cluster config.
+
     Attributes:
         config_bucket (str):
             Optional. A Cloud Storage bucket used to stage job
@@ -228,6 +229,7 @@ class ClusterConfig(proto.Message):
 
 class GkeClusterConfig(proto.Message):
     r"""The GKE config for this cluster.
+
     Attributes:
         namespaced_gke_deployment_target (google.cloud.dataproc_v1.types.GkeClusterConfig.NamespacedGkeDeploymentTarget):
             Optional. A target for the deployment.
@@ -256,6 +258,7 @@ class NamespacedGkeDeploymentTarget(proto.Message):
 
 class EndpointConfig(proto.Message):
     r"""Endpoint config for this cluster
+
     Attributes:
         http_ports (Sequence[google.cloud.dataproc_v1.types.EndpointConfig.HttpPortsEntry]):
             Output only. The map of port descriptions to URLs. Will only
@@ -272,6 +275,7 @@ class EndpointConfig(proto.Message):
 
 class AutoscalingConfig(proto.Message):
     r"""Autoscaling Policy config associated with the cluster.
+
     Attributes:
         policy_uri (str):
             Optional. The autoscaling policy used by the cluster.
@@ -291,6 +295,7 @@ class AutoscalingConfig(proto.Message):
 
 class EncryptionConfig(proto.Message):
     r"""Encryption settings for the cluster.
+
     Attributes:
         gce_pd_kms_key_name (str):
             Optional. The Cloud KMS key name to use for
@@ -687,6 +692,7 @@ class NodeInitializationAction(proto.Message):
 
 class ClusterStatus(proto.Message):
     r"""The status of a cluster and its instances.
+
     Attributes:
         state (google.cloud.dataproc_v1.types.ClusterStatus.State):
             Output only. The cluster's state.
@@ -747,6 +753,7 @@ class SecurityConfig(proto.Message):
 
 class KerberosConfig(proto.Message):
     r"""Specifies Kerberos related configuration.
+
     Attributes:
         enable_kerberos (bool):
             Optional. Flag to indicate whether to
@@ -894,6 +901,7 @@ class SoftwareConfig(proto.Message):
 
 class LifecycleConfig(proto.Message):
     r"""Specifies the cluster auto-delete schedule configuration.
+
     Attributes:
         idle_delete_ttl (google.protobuf.duration_pb2.Duration):
             Optional. The duration to keep the cluster alive while
@@ -935,6 +943,7 @@ class LifecycleConfig(proto.Message):
 
 class MetastoreConfig(proto.Message):
     r"""Specifies a Metastore configuration.
+
     Attributes:
         dataproc_metastore_service (str):
             Required. Resource name of an existing Dataproc Metastore
@@ -967,6 +976,7 @@ class ClusterMetrics(proto.Message):
 
 class CreateClusterRequest(proto.Message):
     r"""A request to create a cluster.
+
     Attributes:
         project_id (str):
             Required. The ID of the Google Cloud Platform
@@ -1001,6 +1011,7 @@ class CreateClusterRequest(proto.Message):
 
 class UpdateClusterRequest(proto.Message):
     r"""A request to update a cluster.
+
     Attributes:
         project_id (str):
             Required. The ID of the Google Cloud Platform
@@ -1117,6 +1128,7 @@ class UpdateClusterRequest(proto.Message):
 
 class StopClusterRequest(proto.Message):
     r"""A request to stop a cluster.
+
     Attributes:
         project_id (str):
             Required. The ID of the Google Cloud Platform
@@ -1156,6 +1168,7 @@ class StopClusterRequest(proto.Message):
 
 class StartClusterRequest(proto.Message):
     r"""A request to start a cluster.
+
     Attributes:
         project_id (str):
             Required. The ID of the Google Cloud Platform
@@ -1195,6 +1208,7 @@ class StartClusterRequest(proto.Message):
 
 class DeleteClusterRequest(proto.Message):
     r"""A request to delete a cluster.
+
     Attributes:
         project_id (str):
             Required. The ID of the Google Cloud Platform
@@ -1254,6 +1268,7 @@ class GetClusterRequest(proto.Message):
 
 class ListClustersRequest(proto.Message):
     r"""A request to list the clusters in a project.
+
     Attributes:
         project_id (str):
             Required. The ID of the Google Cloud Platform
@@ -1298,6 +1313,7 @@ class ListClustersRequest(proto.Message):
 
 class ListClustersResponse(proto.Message):
     r"""The list of all clusters in a project.
+
     Attributes:
         clusters (Sequence[google.cloud.dataproc_v1.types.Cluster]):
             Output only. The clusters in the project.
@@ -1318,6 +1334,7 @@ def raw_page(self):
 
 class DiagnoseClusterRequest(proto.Message):
     r"""A request to collect cluster diagnostic information.
+
     Attributes:
         project_id (str):
             Required. The ID of the Google Cloud Platform
@@ -1336,6 +1353,7 @@ class DiagnoseClusterRequest(proto.Message):
 
 class DiagnoseClusterResults(proto.Message):
     r"""The location of diagnostic output.
+
     Attributes:
         output_uri (str):
             Output only. The Cloud Storage URI of the
@@ -1349,6 +1367,7 @@ class DiagnoseClusterResults(proto.Message):
 
 class ReservationAffinity(proto.Message):
     r"""Reservation Affinity for consuming Zonal reservation.
+
     Attributes:
         consume_reservation_type (google.cloud.dataproc_v1.types.ReservationAffinity.Type):
             Optional. Type of reservation to consume
diff --git a/google/cloud/dataproc_v1/types/jobs.py b/google/cloud/dataproc_v1/types/jobs.py
index 87fa758e..5eece260 100644
--- a/google/cloud/dataproc_v1/types/jobs.py
+++ b/google/cloud/dataproc_v1/types/jobs.py
@@ -52,6 +52,7 @@
 
 class LoggingConfig(proto.Message):
     r"""The runtime logging config of the job.
+
     Attributes:
         driver_log_levels (Sequence[google.cloud.dataproc_v1.types.LoggingConfig.DriverLogLevelsEntry]):
             The per-package log levels for the driver.
@@ -243,6 +244,7 @@ class PySparkJob(proto.Message):
 
 class QueryList(proto.Message):
     r"""A list of queries to run on a cluster.
+
     Attributes:
         queries (Sequence[str]):
             Required. The queries to execute. You do not need to end a
@@ -478,6 +480,7 @@ class PrestoJob(proto.Message):
 
 class JobPlacement(proto.Message):
     r"""Dataproc job config.
+
     Attributes:
         cluster_name (str):
             Required. The name of the cluster where the
@@ -497,6 +500,7 @@ class JobPlacement(proto.Message):
 
 class JobStatus(proto.Message):
     r"""Dataproc job status.
+
     Attributes:
         state (google.cloud.dataproc_v1.types.JobStatus.State):
             Output only. A state message specifying the
@@ -543,6 +547,7 @@ class Substate(proto.Enum):
 
 class JobReference(proto.Message):
     r"""Encapsulates the full scoping used to reference a job.
+
     Attributes:
         project_id (str):
             Optional. The ID of the Google Cloud Platform
@@ -611,6 +616,7 @@ class State(proto.Enum):
 
 class Job(proto.Message):
     r"""A Dataproc job resource.
+
     Attributes:
         reference (google.cloud.dataproc_v1.types.JobReference):
             Optional. The fully qualified reference to the job, which
@@ -716,6 +722,7 @@ class Job(proto.Message):
 
 class JobScheduling(proto.Message):
     r"""Job scheduling options.
+
     Attributes:
         max_failures_per_hour (int):
             Optional. Maximum number of times per hour a
@@ -741,6 +748,7 @@ class JobScheduling(proto.Message):
 
 class SubmitJobRequest(proto.Message):
     r"""A request to submit a job.
+
     Attributes:
         project_id (str):
             Required. The ID of the Google Cloud Platform
@@ -774,6 +782,7 @@ class SubmitJobRequest(proto.Message):
 
 class JobMetadata(proto.Message):
     r"""Job Operation metadata.
+
     Attributes:
         job_id (str):
             Output only. The job id.
@@ -813,6 +822,7 @@ class GetJobRequest(proto.Message):
 
 class ListJobsRequest(proto.Message):
     r"""A request to list jobs in a project.
+
     Attributes:
         project_id (str):
             Required. The ID of the Google Cloud Platform
@@ -873,6 +883,7 @@ class JobStateMatcher(proto.Enum):
 
 class UpdateJobRequest(proto.Message):
     r"""A request to update a job.
+
     Attributes:
         project_id (str):
             Required. The ID of the Google Cloud Platform
@@ -903,6 +914,7 @@ class UpdateJobRequest(proto.Message):
 
 class ListJobsResponse(proto.Message):
     r"""A list of jobs in a project.
+
     Attributes:
         jobs (Sequence[google.cloud.dataproc_v1.types.Job]):
             Output only. Jobs list.
@@ -923,6 +935,7 @@ def raw_page(self):
 
 class CancelJobRequest(proto.Message):
     r"""A request to cancel a job.
+
     Attributes:
         project_id (str):
             Required. The ID of the Google Cloud Platform
@@ -941,6 +954,7 @@ class CancelJobRequest(proto.Message):
 
 class DeleteJobRequest(proto.Message):
     r"""A request to delete a job.
+
     Attributes:
         project_id (str):
             Required. The ID of the Google Cloud Platform
diff --git a/google/cloud/dataproc_v1/types/operations.py b/google/cloud/dataproc_v1/types/operations.py
index 9448110a..bb04624d 100644
--- a/google/cloud/dataproc_v1/types/operations.py
+++ b/google/cloud/dataproc_v1/types/operations.py
@@ -26,6 +26,7 @@
 
 class ClusterOperationStatus(proto.Message):
     r"""The status of the operation.
+
     Attributes:
         state (google.cloud.dataproc_v1.types.ClusterOperationStatus.State):
             Output only. A message containing the
@@ -57,6 +58,7 @@ class State(proto.Enum):
 
 class ClusterOperationMetadata(proto.Message):
     r"""Metadata describing the operation.
+
     Attributes:
         cluster_name (str):
             Output only. Name of the cluster for the
diff --git a/google/cloud/dataproc_v1/types/workflow_templates.py b/google/cloud/dataproc_v1/types/workflow_templates.py
index 98987809..24556467 100644
--- a/google/cloud/dataproc_v1/types/workflow_templates.py
+++ b/google/cloud/dataproc_v1/types/workflow_templates.py
@@ -51,6 +51,7 @@
 
 class WorkflowTemplate(proto.Message):
     r"""A Dataproc workflow template resource.
+
     Attributes:
         id (str):
 
@@ -164,6 +165,7 @@ class WorkflowTemplatePlacement(proto.Message):
 
 class ManagedCluster(proto.Message):
     r"""Cluster that is managed by the workflow.
+
     Attributes:
         cluster_name (str):
             Required. The cluster name prefix. A unique
@@ -218,6 +220,7 @@ class ClusterSelector(proto.Message):
 
 class OrderedJob(proto.Message):
     r"""A job executed by the workflow.
+
     Attributes:
         step_id (str):
             Required. The step id. The id must be unique among all jobs
@@ -385,6 +388,7 @@ class TemplateParameter(proto.Message):
 
 class ParameterValidation(proto.Message):
     r"""Configuration for parameter validation.
+
     Attributes:
         regex (google.cloud.dataproc_v1.types.RegexValidation):
             Validation based on regular expressions.
@@ -402,6 +406,7 @@ class ParameterValidation(proto.Message):
 
 class RegexValidation(proto.Message):
     r"""Validation based on regular expressions.
+
     Attributes:
         regexes (Sequence[str]):
             Required. RE2 regular expressions used to
@@ -415,6 +420,7 @@ class RegexValidation(proto.Message):
 
 class ValueValidation(proto.Message):
     r"""Validation based on a list of allowed values.
+
     Attributes:
         values (Sequence[str]):
             Required. List of allowed values for the
@@ -426,6 +432,7 @@ class ValueValidation(proto.Message):
 
 class WorkflowMetadata(proto.Message):
     r"""A Dataproc workflow template resource.
+
     Attributes:
         template (str):
             Output only. The resource name of the workflow template as
@@ -506,6 +513,7 @@ class State(proto.Enum):
 
 class ClusterOperation(proto.Message):
     r"""The cluster operation triggered by a workflow.
+
     Attributes:
         operation_id (str):
             Output only. The id of the cluster operation.
@@ -522,6 +530,7 @@ class ClusterOperation(proto.Message):
 
 class WorkflowGraph(proto.Message):
     r"""The workflow graph.
+
     Attributes:
         nodes (Sequence[google.cloud.dataproc_v1.types.WorkflowNode]):
             Output only. The workflow nodes.
@@ -532,6 +541,7 @@ class WorkflowGraph(proto.Message):
 
 class WorkflowNode(proto.Message):
     r"""The workflow node.
+
     Attributes:
         step_id (str):
             Output only. The name of the node.
@@ -564,6 +574,7 @@ class NodeState(proto.Enum):
 
 class CreateWorkflowTemplateRequest(proto.Message):
     r"""A request to create a workflow template.
+
     Attributes:
         parent (str):
             Required. The resource name of the region or location, as
@@ -588,6 +599,7 @@ class CreateWorkflowTemplateRequest(proto.Message):
 
 class GetWorkflowTemplateRequest(proto.Message):
     r"""A request to fetch a workflow template.
+
     Attributes:
         name (str):
             Required. The resource name of the workflow template, as
@@ -614,6 +626,7 @@ class GetWorkflowTemplateRequest(proto.Message):
 
 class InstantiateWorkflowTemplateRequest(proto.Message):
     r"""A request to instantiate a workflow template.
+
     Attributes:
         name (str):
             Required. The resource name of the workflow template, as
@@ -661,6 +674,7 @@ class InstantiateWorkflowTemplateRequest(proto.Message):
 
 class InstantiateInlineWorkflowTemplateRequest(proto.Message):
     r"""A request to instantiate an inline workflow template.
+
     Attributes:
         parent (str):
             Required. The resource name of the region or location, as
@@ -699,6 +713,7 @@ class InstantiateInlineWorkflowTemplateRequest(proto.Message):
 
 class UpdateWorkflowTemplateRequest(proto.Message):
     r"""A request to update a workflow template.
+
     Attributes:
         template (google.cloud.dataproc_v1.types.WorkflowTemplate):
             Required. The updated workflow template.
@@ -712,6 +727,7 @@ class UpdateWorkflowTemplateRequest(proto.Message):
 
 class ListWorkflowTemplatesRequest(proto.Message):
     r"""A request to list workflow templates in a project.
+
     Attributes:
         parent (str):
             Required. The resource name of the region or location, as
diff --git a/tests/unit/gapic/dataproc_v1/test_autoscaling_policy_service.py b/tests/unit/gapic/dataproc_v1/test_autoscaling_policy_service.py
index 035bd8d0..880f0c4a 100644
--- a/tests/unit/gapic/dataproc_v1/test_autoscaling_policy_service.py
+++ b/tests/unit/gapic/dataproc_v1/test_autoscaling_policy_service.py
@@ -29,6 +29,7 @@
 from google.api_core import gapic_v1
 from google.api_core import grpc_helpers
 from google.api_core import grpc_helpers_async
+from google.api_core import path_template
 from google.auth import credentials as ga_credentials
 from google.auth.exceptions import MutualTLSChannelError
 from google.cloud.dataproc_v1.services.autoscaling_policy_service import (
@@ -1987,6 +1988,9 @@ def test_autoscaling_policy_service_base_transport():
         with pytest.raises(NotImplementedError):
             getattr(transport, method)(request=object())
 
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
 
 @requires_google_auth_gte_1_25_0
 def test_autoscaling_policy_service_base_transport_with_credentials_file():
@@ -2475,3 +2479,49 @@ def test_client_withDEFAULT_CLIENT_INFO():
             credentials=ga_credentials.AnonymousCredentials(), client_info=client_info,
         )
         prep.assert_called_once_with(client_info)
+
+
+@pytest.mark.asyncio
+async def test_transport_close_async():
+    client = AutoscalingPolicyServiceAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(), transport="grpc_asyncio",
+    )
+    with mock.patch.object(
+        type(getattr(client.transport, "grpc_channel")), "close"
+    ) as close:
+        async with client:
+            close.assert_not_called()
+        close.assert_called_once()
+
+
+def test_transport_close():
+    transports = {
+        "grpc": "_grpc_channel",
+    }
+
+    for transport, close_name in transports.items():
+        client = AutoscalingPolicyServiceClient(
+            credentials=ga_credentials.AnonymousCredentials(), transport=transport
+        )
+        with mock.patch.object(
+            type(getattr(client.transport, close_name)), "close"
+        ) as close:
+            with client:
+                close.assert_not_called()
+            close.assert_called_once()
+
+
+def test_client_ctx():
+    transports = [
+        "grpc",
+    ]
+    for transport in transports:
+        client = AutoscalingPolicyServiceClient(
+            credentials=ga_credentials.AnonymousCredentials(), transport=transport
+        )
+        # Test client calls underlying transport.
+        with mock.patch.object(type(client.transport), "close") as close:
+            close.assert_not_called()
+            with client:
+                pass
+            close.assert_called()
diff --git a/tests/unit/gapic/dataproc_v1/test_cluster_controller.py b/tests/unit/gapic/dataproc_v1/test_cluster_controller.py
index b57aa868..5fae118e 100644
--- a/tests/unit/gapic/dataproc_v1/test_cluster_controller.py
+++ b/tests/unit/gapic/dataproc_v1/test_cluster_controller.py
@@ -32,6 +32,7 @@
 from google.api_core import grpc_helpers_async
 from google.api_core import operation_async  # type: ignore
 from google.api_core import operations_v1
+from google.api_core import path_template
 from google.auth import credentials as ga_credentials
 from google.auth.exceptions import MutualTLSChannelError
 from google.cloud.dataproc_v1.services.cluster_controller import (
@@ -1919,6 +1920,9 @@ def test_cluster_controller_base_transport():
         with pytest.raises(NotImplementedError):
             getattr(transport, method)(request=object())
 
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
     # Additionally, the LRO client (a property) should
     # also raise NotImplementedError
     with pytest.raises(NotImplementedError):
@@ -2456,3 +2460,49 @@ def test_client_withDEFAULT_CLIENT_INFO():
             credentials=ga_credentials.AnonymousCredentials(), client_info=client_info,
         )
         prep.assert_called_once_with(client_info)
+
+
+@pytest.mark.asyncio
+async def test_transport_close_async():
+    client = ClusterControllerAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(), transport="grpc_asyncio",
+    )
+    with mock.patch.object(
+        type(getattr(client.transport, "grpc_channel")), "close"
+    ) as close:
+        async with client:
+            close.assert_not_called()
+        close.assert_called_once()
+
+
+def test_transport_close():
+    transports = {
+        "grpc": "_grpc_channel",
+    }
+
+    for transport, close_name in transports.items():
+        client = ClusterControllerClient(
+            credentials=ga_credentials.AnonymousCredentials(), transport=transport
+        )
+        with mock.patch.object(
+            type(getattr(client.transport, close_name)), "close"
+        ) as close:
+            with client:
+                close.assert_not_called()
+            close.assert_called_once()
+
+
+def test_client_ctx():
+    transports = [
+        "grpc",
+    ]
+    for transport in transports:
+        client = ClusterControllerClient(
+            credentials=ga_credentials.AnonymousCredentials(), transport=transport
+        )
+        # Test client calls underlying transport.
+        with mock.patch.object(type(client.transport), "close") as close:
+            close.assert_not_called()
+            with client:
+                pass
+            close.assert_called()
diff --git a/tests/unit/gapic/dataproc_v1/test_job_controller.py b/tests/unit/gapic/dataproc_v1/test_job_controller.py
index de9fcfc5..05fcd58b 100644
--- a/tests/unit/gapic/dataproc_v1/test_job_controller.py
+++ b/tests/unit/gapic/dataproc_v1/test_job_controller.py
@@ -32,6 +32,7 @@
 from google.api_core import grpc_helpers_async
 from google.api_core import operation_async  # type: ignore
 from google.api_core import operations_v1
+from google.api_core import path_template
 from google.auth import credentials as ga_credentials
 from google.auth.exceptions import MutualTLSChannelError
 from google.cloud.dataproc_v1.services.job_controller import JobControllerAsyncClient
@@ -1837,6 +1838,9 @@ def test_job_controller_base_transport():
         with pytest.raises(NotImplementedError):
             getattr(transport, method)(request=object())
 
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
     # Additionally, the LRO client (a property) should
     # also raise NotImplementedError
     with pytest.raises(NotImplementedError):
@@ -2324,3 +2328,49 @@ def test_client_withDEFAULT_CLIENT_INFO():
             credentials=ga_credentials.AnonymousCredentials(), client_info=client_info,
         )
         prep.assert_called_once_with(client_info)
+
+
+@pytest.mark.asyncio
+async def test_transport_close_async():
+    client = JobControllerAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(), transport="grpc_asyncio",
+    )
+    with mock.patch.object(
+        type(getattr(client.transport, "grpc_channel")), "close"
+    ) as close:
+        async with client:
+            close.assert_not_called()
+        close.assert_called_once()
+
+
+def test_transport_close():
+    transports = {
+        "grpc": "_grpc_channel",
+    }
+
+    for transport, close_name in transports.items():
+        client = JobControllerClient(
+            credentials=ga_credentials.AnonymousCredentials(), transport=transport
+        )
+        with mock.patch.object(
+            type(getattr(client.transport, close_name)), "close"
+        ) as close:
+            with client:
+                close.assert_not_called()
+            close.assert_called_once()
+
+
+def test_client_ctx():
+    transports = [
+        "grpc",
+    ]
+    for transport in transports:
+        client = JobControllerClient(
+            credentials=ga_credentials.AnonymousCredentials(), transport=transport
+        )
+        # Test client calls underlying transport.
+        with mock.patch.object(type(client.transport), "close") as close:
+            close.assert_not_called()
+            with client:
+                pass
+            close.assert_called()
diff --git a/tests/unit/gapic/dataproc_v1/test_workflow_template_service.py b/tests/unit/gapic/dataproc_v1/test_workflow_template_service.py
index ed87a04e..3e220b7a 100644
--- a/tests/unit/gapic/dataproc_v1/test_workflow_template_service.py
+++ b/tests/unit/gapic/dataproc_v1/test_workflow_template_service.py
@@ -32,6 +32,7 @@
 from google.api_core import grpc_helpers_async
 from google.api_core import operation_async  # type: ignore
 from google.api_core import operations_v1
+from google.api_core import path_template
 from google.auth import credentials as ga_credentials
 from google.auth.exceptions import MutualTLSChannelError
 from google.cloud.dataproc_v1.services.workflow_template_service import (
@@ -2450,6 +2451,9 @@ def test_workflow_template_service_base_transport():
         with pytest.raises(NotImplementedError):
             getattr(transport, method)(request=object())
 
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
     # Additionally, the LRO client (a property) should
     # also raise NotImplementedError
     with pytest.raises(NotImplementedError):
@@ -3017,3 +3021,49 @@ def test_client_withDEFAULT_CLIENT_INFO():
             credentials=ga_credentials.AnonymousCredentials(), client_info=client_info,
         )
         prep.assert_called_once_with(client_info)
+
+
+@pytest.mark.asyncio
+async def test_transport_close_async():
+    client = WorkflowTemplateServiceAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(), transport="grpc_asyncio",
+    )
+    with mock.patch.object(
+        type(getattr(client.transport, "grpc_channel")), "close"
+    ) as close:
+        async with client:
+            close.assert_not_called()
+        close.assert_called_once()
+
+
+def test_transport_close():
+    transports = {
+        "grpc": "_grpc_channel",
+    }
+
+    for transport, close_name in transports.items():
+        client = WorkflowTemplateServiceClient(
+            credentials=ga_credentials.AnonymousCredentials(), transport=transport
+        )
+        with mock.patch.object(
+            type(getattr(client.transport, close_name)), "close"
+        ) as close:
+            with client:
+                close.assert_not_called()
+            close.assert_called_once()
+
+
+def test_client_ctx():
+    transports = [
+        "grpc",
+    ]
+    for transport in transports:
+        client = WorkflowTemplateServiceClient(
+            credentials=ga_credentials.AnonymousCredentials(), transport=transport
+        )
+        # Test client calls underlying transport.
+        with mock.patch.object(type(client.transport), "close") as close:
+            close.assert_not_called()
+            with client:
+                pass
+            close.assert_called()