private void deployJob() throws KeeperException, InterruptedException, IOException {
        this.jobDesc = (JobDescriptor) zookeeper.getObject(ZookeeperStrings.jobDescPath(jobId), this, null);
        final GlobalPhysicalGraph graph
                = (GlobalPhysicalGraph) zookeeper.getObject(ZookeeperStrings.jobScheduledGraphPath(jobId), false, null);
        final LocalPhysicalGraph lpg = worker.deployJob(jobDesc, graph);

        Iterator<LocalVertex> iter = lpg.iteratorOfVertex();
        HashSet<Integer> done = new HashSet<>();
        while (iter.hasNext()) {
            final LocalVertex lv = iter.next();

            if (lv instanceof LocalPhysicalVertex) {
                final LocalPhysicalVertex lpv = (LocalPhysicalVertex) lv;
                if (done.contains(lpv.getGlobalVertexId())) {
                    continue;
                }

                done.add(lpv.getGlobalVertexId());

                VertexRunningInformation info = new VertexRunningInformation(lpv.getGlobalVertexId(),
                                                                             worker.getInetAddress());

                for (int i = 0; i < lpv.getInputQueueCount(); i++) {
                    if (lpv.hasRemoteSource(i)) {
                        final LocalEdge selectorEdge = lpv.getSourceSelectorEdge(i);
                        final LocalSelectorVertex selectorVertex = (LocalSelectorVertex) selectorEdge.getSrcVertex();
                        info.addAcceptAddress(i, selectorVertex.getAcceptAddress(selectorEdge.getSrcQueueId()));
                        final LocalEdge edge = lpv.getInputEdges(i).get(0);
                        final GlobalPhysicalEdge gpe = edge.getGlobalPhysicalEdge();

                        assert (gpe.getLogicalEdge().getDestQueueHandle().getQueueId() == i);

//                        System.out.printf("A vertex %d (queue %d) will connect to a vertex %d (queue %d) on %s\n",
//                                          gpe.getSrcGlobalVertex().getGlobalPhysicalVertexId(),
//                                          gpe.getSrcQueueLogicalId(),
//                                          lpv.getGlobalVertexId(), i,
//                                          selectorVertex.getAcceptAddress(selectorEdge.getSrcQueueId()).toString());
                    }
                }

                zookeeper.create(ZookeeperStrings.jobVertexInfoPath(jobId, lpv.getGlobalVertexId()),
                                 info, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        }

        checkStatus();
    }