public void initialExecute() throws InterruptedException, KeeperException {
        jobId = zookeeper.getAndIncrementAtomicInt(ZookeeperStrings.nextJobIdPath());

        if (jobId != job.getJobId()) {
            throw new AssertionError(String.format("BUG: The job ID is inconsistent. %d %d", jobId, job.getJobId()));
        }

        zookeeper.create(ZookeeperStrings.jobPath(jobId), null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zookeeper.createAtomicInt(ZookeeperStrings.jobReadyCountdownLatch(jobId),
                                  Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zookeeper.create(ZookeeperStrings.jobVerticesPath(jobId), null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zookeeper.create(ZookeeperStrings.jobScheduledGraphPath(jobId),
                         graph, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zookeeper.create(ZookeeperStrings.jobStatusPath(jobId), jobStatus, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zookeeper.create(ZookeeperStrings.jobDescPath(jobId), job, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zookeeper.create(ZookeeperStrings.jobStatsTopPath(jobId), null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        Collection<WorkerSignature> workers = graph.getAllWorkers();
        for (WorkerSignature worker : workers) {
            zookeeper.create(ZookeeperStrings.jobStatsPath(jobId, worker.getServerId()),
                             null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

        List<String> children = zookeeper.getChildren(ZookeeperStrings.jobVerticesPath(jobId), this);
        if (children.size() != 0) {
            execute();
        }
    }