@Override
    public ReduceUnit next() {

        List<String> values = new ArrayList<String>();

        RecordReaderWrapper rrw = priorityQueue.poll();
        String key = rrw.record.getKey();
        values.add(rrw.record.getValue());

        //System.out.println("next Key:" + key);

        while(true){

            try {
                RecordReaderWrapper rrw2 = moveNext(rrw.reader);
                if(rrw2 != null){
                    priorityQueue.add(rrw2);
                }
            } catch (IOException e) {
                System.out.println("Error occur when adding item!");
            }

            RecordReaderWrapper peek = priorityQueue.peek();
            if(peek == null) break;

            String nextKey = peek.record.getKey();
            if(!nextKey.equals(key)) break;
            else{
                rrw = priorityQueue.poll();
                values.add(rrw.record.getValue());
            }
        }

        return new ReduceUnit(key,values);
    }