

@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
    final_countdown_timer_secs = 50.0
    sleep_secs = 5.0

    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 (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'):
                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
                sleep_secs = 1.0
                final_countdown_timer_secs -= sleep_secs
        time.sleep(sleep_secs)
        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)):
                    continue
                last_message_time = m.time
                last_message_id = m.id
                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