@staticmethod
  def poll_for_job_completion(runner, job_id):
    """Polls for the specified job to finish running (successfully or not)."""
    last_message_time = None
    last_message_id = None

    last_error_rank = float('-inf')
    last_error_msg = None
    last_job_state = None
    # How long to wait after pipeline failure for the error
    # message to show up giving the reason for the failure.
    # It typically takes about 30 seconds.
    final_countdown_timer_secs = 50.0
    sleep_secs = 5.0
    # Try to prioritize the user-level traceback, if any.
    def rank_error(msg):
      if 'work item was attempted' in msg:
        return -1
      elif 'Traceback' in msg:
        return 1
      else:
        return 0

    while True:
      response = runner.dataflow_client.get_job(job_id)
      # If get() is called very soon after Create() the response may not contain
      # an initialized 'currentState' field.
      if response.currentState is not None:
        if response.currentState != last_job_state:
          logging.info('Job %s is in state %s', job_id, response.currentState)
          last_job_state = response.currentState
        if str(response.currentState) != 'JOB_STATE_RUNNING':
          # Stop checking for new messages on timeout, explanatory
          # message received, success, or a terminal job state caused
          # by the user that therefore doesn't require explanation.
          if (final_countdown_timer_secs <= 0.0
              or last_error_msg is not None
              or str(response.currentState) == 'JOB_STATE_DONE'
              or str(response.currentState) == 'JOB_STATE_CANCELLED'
              or str(response.currentState) == 'JOB_STATE_UPDATED'
              or str(response.currentState) == 'JOB_STATE_DRAINED'):
            break
          # The job has failed; ensure we see any final error messages.
          sleep_secs = 1.0      # poll faster during the final countdown
          final_countdown_timer_secs -= sleep_secs
      time.sleep(sleep_secs)

      # Get all messages since beginning of the job run or since last message.
      page_token = None
      while True:
        messages, page_token = runner.dataflow_client.list_messages(
            job_id, page_token=page_token, start_time=last_message_time)
        for m in messages:
          if last_message_id is not None and m.id == last_message_id:
            # Skip the first message if it is the last message we got in the
            # previous round. This can happen because we use the
            # last_message_time as a parameter of the query for new messages.
            continue
          last_message_time = m.time
          last_message_id = m.id
          # Skip empty messages.
          if m.messageImportance is None:
            continue
          logging.info(
              '%s: %s: %s: %s', m.id, m.time, m.messageImportance,
              m.messageText)
          if str(m.messageImportance) == 'JOB_MESSAGE_ERROR':
            if rank_error(m.messageText) >= last_error_rank:
              last_error_rank = rank_error(m.messageText)
              last_error_msg = m.messageText
        if not page_token:
          break

    runner.result = DataflowPipelineResult(response)
    runner.last_error_msg = last_error_msg
dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/runners/dataflow_runner.py/DataflowPipelineRunner.poll_for_job_completion/original