Repository: oppia/oppia
File: jobs/jobs_manager.py (Python)

Code before:
    """
    job_id = beam_job_run_model.dataflow_job_id
    if job_id is None:
        beam_job_run_model.latest_job_state = (
            beam_job_models.BeamJobState.UNKNOWN.value)
        beam_job_run_model.update_timestamps(update_last_updated_time=False)
        return

    try:
        job = dataflow.JobsV1Beta3Client().get_job(dataflow.GetJobRequest(
            job_id=job_id, project_id=feconf.OPPIA_PROJECT_ID,
            location=feconf.GOOGLE_APP_ENGINE_REGION))

    except Exception:
        job_state = beam_job_models.BeamJobState.UNKNOWN.value
        job_state_updated = beam_job_run_model.last_updated

        logging.exception('Failed to update job_id="%s"!' % job_id)

    else:
        job_state = _GCLOUD_DATAFLOW_JOB_STATE_TO_OPPIA_BEAM_JOB_STATE.get(
            job.current_state, beam_job_models.BeamJobState.UNKNOWN).value
        job_state_updated = job.current_state_time.replace(tzinfo=None)

        if (beam_job_run_model.latest_job_state != job_state and
                job_state == beam_job_models.BeamJobState.FAILED.value):
            _put_job_stderr(beam_job_run_model.id, pprint.pformat(job))

    beam_job_run_model.latest_job_state = job_state
    beam_job_run_model.last_updated = job_state_updated
    beam_job_run_model.update_timestamps(update_last_updated_time=False)


@contextlib.contextmanager
def _job_bookkeeping_context(
    job_name: str
) -> Iterator[beam_job_models.BeamJobRunModel]:
    """Returns a context manager which commits failure details if an exception
    occurs.

    Args:
        job_name: str. The name of the job.

    Yields:
        BeamJobRunModel. The bookkeeping model used to record execution details.
    """
    run_model = beam_job_services.create_beam_job_run_model(job_name)

    try:
        yield run_model


Review comment (bug):
```suggestion
        if (
                beam_job_run_model.latest_job_state != job_state and
                job_state == beam_job_models.BeamJobState.FAILED.value
        ):
```

Code after:
    """
    job_id = beam_job_run_model.dataflow_job_id
    if job_id is None:
        beam_job_run_model.latest_job_state = (
            beam_job_models.BeamJobState.UNKNOWN.value)
        beam_job_run_model.update_timestamps(update_last_updated_time=False)
        return

    try:
        job = dataflow.JobsV1Beta3Client().get_job(dataflow.GetJobRequest(
            job_id=job_id, project_id=feconf.OPPIA_PROJECT_ID,
            location=feconf.GOOGLE_APP_ENGINE_REGION))

    except Exception:
        job_state = beam_job_models.BeamJobState.UNKNOWN.value
        job_state_updated = beam_job_run_model.last_updated

        logging.exception('Failed to update job_id="%s"!' % job_id)

    else:
        job_state = _GCLOUD_DATAFLOW_JOB_STATE_TO_OPPIA_BEAM_JOB_STATE.get(
            job.current_state, beam_job_models.BeamJobState.UNKNOWN).value
        job_state_updated = job.current_state_time.replace(tzinfo=None)

        if (
                beam_job_run_model.latest_job_state != job_state and
                job_state == beam_job_models.BeamJobState.FAILED.value
        ):
            _put_job_stderr(beam_job_run_model.id, pprint.pformat(job))

    beam_job_run_model.latest_job_state = job_state
    beam_job_run_model.last_updated = job_state_updated
    beam_job_run_model.update_timestamps(update_last_updated_time=False)


@contextlib.contextmanager
def _job_bookkeeping_context(
    job_name: str
) -> Iterator[beam_job_models.BeamJobRunModel]:
    """Returns a context manager which commits failure details if an exception
    occurs.

    Args:
        job_name: str. The name of the job.

    Yields:
        BeamJobRunModel. The bookkeeping model used to record execution details.
    """
    run_model = beam_job_services.create_beam_job_run_model(job_name)

    try: