As soon as run() is called in the RTR, a task goes into pending. Tasks go into running state as soon as we find a viable worker to run the task (this wil be before the task is actually started on the worker). When tasks fail, they remain in the running state until after some retry delay, they then go into the pending queue again. Each time run() is called, or if a worker completes a task or if a new worker is added to the cluster, we will try to pull out all tasks that are pending and run them. Tasks are run by submitting a runnable to a single threaded executor. This should prevent different places trying to assign tasks and clobbering each other. Is this what you had in mind?