code: public void reduce(K2 key, Iterator<V2> values, 
                       OutputCollector<K3, V3> output, Reporter reporter
                       ) throws IOException {
      isOk = false;
      startApplication(output, reporter);
      downlink.reduceKey(key);
      while (values.hasNext()) {
        downlink.reduceValue(values.next());
      }
      if(skipping) {
        //flush the streams on every record input if running in skip mode
        //so that we don't buffer other records surrounding a bad record.
        downlink.flush();
      }
      isOk = true;
    }
docs: Process all of the keys and values. Start up the application if we haven't
  started it yet.