job flows are not reused when the previous job failed

[garrettjacobson]

Running a simple mrjob that raises an exception in the input mapper causes that job flow to not be re-usable automatically on the next run. It is still shown as WAITING, even though it is not automatically re-used. If you remove the exception and the job succeeds, the jobflow is also shown as WAITING but it will be reused on subsequent runs.

[ap-ensighten]

I also experienced this with the job flow pooling options. However, I was able to continue to use the cluster by specifying its jobflow id directly with --emr-job-flow-id. I think this has to do with the locking mechanism mrjob uses with files on s3, but I have not had a chance to dig into the code yet. I tried deleting the lock file in the s3 lock directory and running another job with the pooling options, but it still would not re-use the jobflow.

[coyotemarin]

This might pair well with other pooling tickets: #707, #708.

[coyotemarin]

I think this is the culprit (from usable_job_flows() in mrjob/emr.py):

            # in rare cases, job flow can be WAITING *and* have incomplete
            # steps
            if any(getattr(step, 'enddatetime', None) is None
                   for step in job_flow.steps):
                return

I think we just need to be pickier about only applying this to steps that have not yet run.

[coyotemarin]

Whoops, found a bug involving an interaction between pooling and max_hours_idle. Fixing now.

[coyotemarin]

Aha! The problem is that if you have a multi-step job, and the first step fails, the rest of the steps go into the CANCELLED state (because their input doesn't exist) and, because they've never run, their enddatetime is None.