public JobStatistics getJobStatistics(final JobDescriptor jobDesc) {
        final GlobalPhysicalGraph graph = jobManager.getPhysicalGraph(jobDesc);
        Collection<WorkerSignature> workers = graph.getAllWorkers();
        List<LocalJobStatistics> stats = server.getStatistics(jobDesc, workers);

        JobStatistics ret = new JobStatistics();
        final LogicalGraph lg = jobDesc.getJob().getVertexGraph();
        Iterator<Vertex> iter = lg.iteratorOfVertex();
        while (iter.hasNext()) {
            final Vertex v = iter.next();
            // Aggregate the statistics.
            long contextCount = 0, threadCount = 0, inputCount = 0, outputCount = 0, consumedMemory = 0;
            for (LocalJobStatistics stat : stats) {
                List<LocalVertexStat> lvss = stat.getLocalVertexStat(v.getVertexId());
                if (lvss == null) {
                    continue;
                }
                for (LocalVertexStat lvs : lvss) {
                    contextCount += lvs.getWaitingOnThread();
                    threadCount += lvs.getWaitingOnThread();
                    inputCount += lvs.getInputCount();
                    outputCount += lvs.getOutputCount();
                    consumedMemory += lvs.getConsumedMemory();
                }
            }

            ret.addStat(v, contextCount, threadCount, inputCount, outputCount, consumedMemory);
        }

        return ret;
    }