[FLINK-5501] JM use running job registry to determine whether is the first running

[shuai-xu]

This pr if for jira-#5501.

The main changes are:

1. Add interface isJobFinished() and clearJob() to RunningJobRegistry and implement them.
2. After grantLeadership, JMRunner will first check whether the job is finished, if finished, it means that other JM has finished the job, it only need to exist.
3. Then JMRunner will check whether the job is running, if running, it means other JM has run it, but not succeeded, so it need to recover it.
4. If the job is not running, it mean the first running, the JMRunner will setJobRunning in RunningJobRegistry.
5. After job finished, will clear the job state from RunningJobRegistry

[StephanEwen]

PR looks like a good start, but I think we need to add a few things on top:

• The file-based registry cannot distinguish between "job created but not running" and "job running". This distinction is important to decide whether to start reconciliation.
• There are currently no tests for the extended functionality of the file-based registry

[StephanEwen]

I would like to merge this and make a few edits on top...

[StephanEwen]

One issue I think can happen in practice is that the checks "isRunning" and "isFinished" are not atomic. Imagine this scenario:

• job is running
• JobManager checks "isFinished" -> false
• job finishes
• JobManager checks "isRunning" -> false
• JobManager starts job = bug

[StephanEwen]

With the problem observed above, I think we should change the approach a bit:

• The registry should have an enum that it returns: getJobSchedulingStatus or so, which can be PENDING, RUNNING, and DONE. That way there is only one access to the registry and we don't have the problem that the internal status is changed between checks.

• The file-based registry would create one file for the transition to RUNNING and another for the transition to DONE. Important is that the transition to DONE does not remove the file for RUNNING. The status check checks backwards - first for the DONE file, then for the RUNNING file.

[StephanEwen]

String to bytes conversion (and bytes to string) must always explicitly specify the encoding (Charset). Otherwise, there can be mismatches when different machines configure different default Charsets.

[shuai-xu]

hi @StephanEwen , thank for you review, I modify it according to your comments, add getJobSchedulingStatus to it and add tests.

[StephanEwen]

Thanks!
I think I can take this over now...

[StephanEwen]

One test case seemed to be failing in this PR:
I have merged the PR to my local repository, fixed the test, and added some fixes/cleanups on top.
Will merge back to Flink master tomorrow...

[shuai-xu]

@StephanEwen , Thank you very much, sorry for the test break, next time I will be more careful.