目前 Dispatcher 在收到 JobManagerRunner 完成 Job 的消息后会去调用 JobManagerRunner#closeAsync，随后清理 SubmittedJobGraphStore
在不考虑 RunningJobsRegistry 的情况下的由于这个过程不是原子的，所以有可能

1. 在清理 SubmittedJobGraphStore 之前 Dispatcher 挂了，重启的 Dispatcher 从 SubmittedJobGraphStore 中恢复 JobGraph，又去执行已经完成的 Job
2. 在 JobManagerRunner#closeAsync 过程中，原来的 JobManagerRunner 放弃 leadership，但是还未关闭 LeaderElectionService，此时新的 JobManagerRunner 起来又去执行已经完成的 Job

目前通过 JobManagerRunner 在完成任务时往 RunningJobsRegistry 写 JobSchedulingStatus#DONE 来拦住这两个行为，但是 RunningJobsRegistry 最终却是由 Dispatcher 来清理的，这样就把一个数据结构的管理职责分散给了两个角色。究其原因是因为 Job 的状态原本应该由 Dispatcher 来 commit 和发布，现在却以 JobManagerRunner 的状态为准。解决这种职责混乱的方法可以是让 Dispatcher 来负责 Job 状态的发布，并在 Dispatcher 端原子化 Job 的 commit 和 SubmittedJobGraphStore 的清理工作，同时在此时阻止新的 JobId 相同的 JobManagerRunner 的启动

(tison, 2018-12-09)