class_docstr: Check for the state of batch.
  
  :param batch_id: The Dataproc batch ID to poll. (templated)
  :param region: Required. The Cloud Dataproc region in which to handle the request. (templated)
  :param project_id: The ID of the google cloud project in which
      to create the cluster. (templated)
  :param gcp_conn_id: The connection ID to use connecting to Google Cloud Platform.
  :param wait_timeout: How many seconds wait for job to be ready.
class_docstr_tok: ["Check","for","the","state","of","batch",".",":","param","batch_id",":","The","Dataproc","batch","ID","to","poll",".","(","templated",")",":","param","region",":","Required",".","The","Cloud","Dataproc","region","in","which","to","handle","the","request",".","(","templated",")",":","param","project_id",":","The","ID","of","the","google","cloud","project","in","which","to","create","the","cluster",".","(","templated",")",":","param","gcp_conn_id",":","The","connection","ID","to","use","connecting","to","Google","Cloud","Platform",".",":","param","wait_timeout",":","How","many","seconds","wait","for","job","to","be","ready","."]
code: def poke(self, context: Context) -> bool:
      hook = DataprocHook(gcp_conn_id=self.gcp_conn_id)
      if self.wait_timeout:
          try:
              batch = hook.get_batch(
                  batch_id=self.batch_id,
                  region=self.region,
                  project_id=self.project_id,
              )
          except ServerError as err:
              duration = self._duration()
              self.log.info("DURATION RUN: %f", duration)
  
              if duration > self.wait_timeout:
                  raise AirflowException(
                      f"Timeout: dataproc batch {self.batch_id} is not ready after {self.wait_timeout}s"
                  )
              self.log.info(
                  "Retrying. Dataproc API returned server error when waiting for batch: %s",
                  err,
              )
              return False
      else:
          batch = hook.get_batch(
              batch_id=self.batch_id,
              region=self.region,
              project_id=self.project_id,
          )
  
      state = batch.state
      if state == Batch.State.FAILED:
          # TODO: remove this if check when min_airflow_version is set to higher than 2.7.1
          message = "Batch failed"
          if self.soft_fail:
              raise AirflowSkipException(message)
          raise AirflowException(message)
      elif state in {
          Batch.State.CANCELLED,
          Batch.State.CANCELLING,
      }:
          # TODO: remove this if check when min_airflow_version is set to higher than 2.7.1
          message = "Batch was cancelled."
          if self.soft_fail:
              raise AirflowSkipException(message)
          raise AirflowException(message)
      elif state == Batch.State.SUCCEEDED:
          self.log.debug(
              "Batch %s completed successfully.", self.batch_id
          )
          return True
  
      self.log.info(
          "Waiting for the batch %s to complete.", self.batch_id
      )
      return False
  
file_path: airflow/airflow/providers/google/cloud/sensors/dataproc.py
imports: from __future__ import annotations
  import time
  from typing import TYPE_CHECKING, Sequence
  from google.api_core.exceptions import ServerError
  from google.cloud.dataproc_v1.types import Batch, JobStatus
  from airflow.exceptions import AirflowException, AirflowSkipException
  from airflow.providers.google.cloud.hooks.dataproc import DataprocHook
  from airflow.sensors.base import BaseSensorOperator
label_desc: Use image node_id 4 for calling the DataprocBatchSensor obj's underlying member method code with example usage: obj.poke(context) and returns: False, False, True