@Override
    public void updateStats(final PhysicalGraph graph) throws ClientException {
        List<WorkerSignature> workers = getAllWorkers();
        HashMap<Integer, LocalJobStatistics> statsMap = new HashMap<>();
        for (WorkerSignature worker : workers) {
            try {
                final LocalJobStatistics localStat = (LocalJobStatistics) zookeeper.getObject(ZookeeperStrings
                        .jobStatsPath(graph.getJobId(), worker.getServerId()), false, null);
                if (localStat == null) {
                    return;
                }
                statsMap.put(worker.getServerId(), localStat);
            } catch (KeeperException | InterruptedException e) {
                throw new ClientException(e);
            }
        }

        Iterator<PhysicalVertex> iter = graph.iteratorOfVertex();
        while (iter.hasNext()) {
            final PhysicalVertex pv = iter.next();
            final LocalJobStatistics stats = statsMap.get(pv.getWorkerSignature().getServerId());
            final List<LocalVertexStat> list = stats.getLocalVertexStat(pv.getPhysicalId());
            long inputCount = 0, outputCount = 0;
            for (LocalVertexStat stat : list) {
                inputCount += stat.getInputCount();
                outputCount += stat.getOutputCount();
            }
            pv.setInputCountStat(inputCount);
            pv.setOutputCountStat(outputCount);
        }
    }