public void startConnecting(final JobDescriptor jobDesc) throws IOException {
        LocalPhysicalGraph lpg = jobManager.getLocalPhysicalGraph(jobDesc);
        Iterator<LocalVertex> iter = lpg.iteratorOfVertex();
        while (iter.hasNext()) {
            LocalVertex lv = iter.next();
            if (lv instanceof LocalSelectorVertex) {
                final LocalSelectorVertex lsv = (LocalSelectorVertex) lv;
                final SelectorContext<?, ?> context = (SelectorContext<?, ?>) lsv.getContext();

                for (int i = 0; i < lsv.getInputQueueCount(); i++) {
                    final List<LocalEdge> edges = lsv.getInputEdges(i);
                    final LocalEdge edge = edges.get(0);
                    final GlobalPhysicalEdge gpe = edge.getGlobalPhysicalEdge();
                    final SocketAddress address = workerServer.getRemoteAddress(jobDesc.getJobId(), gpe
                            .getDestGlobalVertex().getGlobalPhysicalVertexId(), gpe.getLogicalEdge()
                            .getDestQueueHandle().getQueueId());

                    context.connect(i, address);

                    // log ("A vertex %d (queue %d) is connecting to a vertex %d (queue %d) on %s.\n",
                    // gpe.getSrcGlobalVertex()
                    // .getGlobalPhysicalVertexId(), gpe
                    // .getLogicalEdge()
                    // .getSrcQueueHandle().getQueueId(),
                    // gpe.getDestGlobalVertex()
                    // .getGlobalPhysicalVertexId(), gpe
                    // .getLogicalEdge()
                    // .getDestQueueHandle().getQueueId(),
                    // address);
                }
            }
        }
    }