@Override
    public void compute(Vertex<Text, Text, NullWritable> vertex, Iterable<Text> messages) throws IOException {
        if (getSuperstep() == MAX_SIZE.get(getConf())) {
            vertex.voteToHalt();
            return;
        }

        if (getSuperstep() == 0) {
            doFirstStep(vertex);
        } else {
            doStep(vertex, messages);
        }
        vertex.voteToHalt();
    }