code: @Override
  	public final void reduce(Iterable<IN> values, Collector<OUT> out) throws Exception {
  		for (Iterator<OUT> iter = reduceGroup(values); iter.hasNext(); ) {
  			out.collect(iter.next());
  		}
  	}
docs: -------------------------------------------------------------------------------------------