class_docstr: Google Cloud Dataproc APIs.
  
  All the methods in the hook where project_id is used must be called with
  keyword arguments rather than positional.
class_docstr_tok: ["Google","Cloud","Dataproc","APIs",".","All","the","methods","in","the","hook","where","project_id","is","used","must","be","called","with","keyword","arguments","rather","than","positional","."]
docstr: Poll a job to check if it has finished.
  
  :param job_id: Dataproc job ID.
  :param project_id: Google Cloud project ID that the cluster belongs to.
  :param region: Cloud Dataproc region to handle the request.
  :param wait_time: Number of seconds between checks.
  :param timeout: How many seconds wait for job to be ready.
code: def wait_for_job(
      self,
      job_id: str,
      project_id: str,
      region: str,
      wait_time: int = 10,
      timeout: int | None = None,
  ) -> None:
      """Poll a job to check if it has finished.
  
      :param job_id: Dataproc job ID.
      :param project_id: Google Cloud project ID that the cluster belongs to.
      :param region: Cloud Dataproc region to handle the request.
      :param wait_time: Number of seconds between checks.
      :param timeout: How many seconds wait for job to be ready.
      """
      if region is None:
          raise TypeError(
              "missing 1 required keyword argument: 'region'"
          )
      state = None
      start = time.monotonic()
      while state not in (
          JobStatus.State.ERROR,
          JobStatus.State.DONE,
          JobStatus.State.CANCELLED,
      ):
          if timeout and start + timeout < time.monotonic():
              raise AirflowException(
                  f"Timeout: dataproc job {job_id} is not ready after {timeout}s"
              )
          time.sleep(wait_time)
          try:
              job = self.get_job(
                  project_id=project_id, region=region, job_id=job_id
              )
              state = job.status.state
          except ServerError as err:
              self.log.info(
                  "Retrying. Dataproc API returned server error when waiting for job: %s",
                  err,
              )
  
      if state == JobStatus.State.ERROR:
          raise AirflowException(f"Job failed:\n{job}")
      if state == JobStatus.State.CANCELLED:
          raise AirflowException(f"Job was cancelled:\n{job}")
  
imports: from __future__ import annotations
  import time
  import uuid
  from typing import TYPE_CHECKING, Any, Sequence
  from google.api_core.client_options import ClientOptions
  from google.api_core.exceptions import ServerError
  from google.api_core.gapic_v1.method import DEFAULT, _MethodDefault
  from google.cloud.dataproc_v1 import Batch, BatchControllerAsyncClient, BatchControllerClient, Cluster, ClusterControllerAsyncClient, ClusterControllerClient, Job, JobControllerAsyncClient, JobControllerClient, JobStatus, WorkflowTemplate, WorkflowTemplateServiceAsyncClient, WorkflowTemplateServiceClient
  from airflow.exceptions import AirflowException
  from airflow.providers.google.common.consts import CLIENT_INFO
  from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
  from airflow.version import version
label_desc: Use image node_id 17 for calling the DataprocHook obj's underlying member method code with example usage: obj.wait_for_job(job_id, project_id, region, wait_time, timeout) without return types