private void checkStatus() throws KeeperException, InterruptedException, IOException {
        if (zookeeper.exists(ZookeeperStrings.jobStatusPath(jobId), this) == null) {
            return;
        }

        final JobStatus status = (JobStatus) zookeeper.getObject(ZookeeperStrings.jobStatusPath(jobId), this, null);
        if (status.getStatus() == Status.CONNECTING) {
            worker.startConnecting(jobDesc);
            zookeeper.getAndIncrementAtomicInt(ZookeeperStrings.jobReadyCountdownLatch(jobId));
        } else if (status.getStatus() == Status.RUNNING) {
            worker.startVertices(jobDesc);
        } else if (status.getStatus() == Status.TERMINATING) {
            worker.stopJob(jobDesc);
        }
    }