public void handleEvents(List<PrioritizableStageDriver> drivers, List<StageProcessTimingEvent> events) {
        if (events.isEmpty()) {
            return;
        }

        Map<Stage, Tuple> timings = new HashMap<Stage, Tuple>();
        long total = 0;
        for (StageProcessTimingEvent ev : events) {
            Tuple tuple = timings.get((Stage) ev.getSource());
            if (tuple == null) {
                tuple = new Tuple();
                timings.put((Stage) ev.getSource(), tuple);
            }

            tuple.add(ev.getLatency());
            total += ev.getLatency();
        }

        if (log.isDebugEnabled()) {
            log.debug("Events handled: " + events.size());
            log.debug("Stage latencies: ");
            for (Map.Entry<Stage, Tuple> entry : timings.entrySet()) {
                log.debug(entry.getKey() + ": " + entry.getValue().duration / entry.getValue().count + "; ");
            }
            log.debug("Total latency: " + total);
        }

        double mean = total / events.size();
        //log.debug("Mean latency: " + mean);

        for (PrioritizableStageDriver driver : drivers) {
            Tuple tuple = timings.get(driver.getStage());
            if (tuple != null) {
                long averageDuration = tuple.duration / tuple.count;
                if (averageDuration > mean + allowableDelta) {
                    log.debug("Increasing priority for stage " + driver.getStage() + " with average duration " + averageDuration);
                    driver.increasePriority(1);
                } else if (averageDuration < mean - allowableDelta) {
                    driver.decreasePriority(1);
                    log.debug("Decreasing priority for stage " + driver.getStage() + " with average duration " + averageDuration);
                }
            }
        }
    }