/**
 * Copyright (c) 2011, Cloudera, Inc. All Rights Reserved.
 *
 * Cloudera, Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"). You may not use this file except in
 * compliance with the License. You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
 * CONDITIONS OF ANY KIND, either express or implied. See the License for
 * the specific language governing permissions and limitations under the
 * License.
 */
package com.cloudera.crunch.impl.mr.run;

import java.io.IOException;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapreduce.Reducer;

public class CrunchReducer extends Reducer<Object, Object, Object, Object> {

  private static final Log LOG = LogFactory.getLog(CrunchReducer.class);
  
  private List<RTNode> nodes;
  private CrunchTaskContext ctxt;

  @Override
  protected void setup(Reducer<Object, Object, Object, Object>.Context context) {
    RTNodeSerializer serde = new RTNodeSerializer();
    this.ctxt = new CrunchTaskContext(context, NodeContext.REDUCE);
    try {
      nodes = serde.deserialize(ctxt);
    } catch (IOException e) {
      LOG.info("Crunch deserialization error", e);
      throw new CrunchRuntimeException(e);
    }
  }

  @Override
  protected void reduce(Object key, Iterable<Object> values,
      Reducer<Object, Object, Object, Object>.Context context) {
    for (RTNode node : nodes) {
      node.process(key, values);
    }
  }

  @Override
  protected void cleanup(Reducer<Object, Object, Object, Object>.Context context) {
    for (RTNode node : nodes) {
      node.cleanup();
    }
    ctxt.cleanup();
  }
}

@@ -955,26 +955,19 @@
 ate 
-List%3C
 RTNode
-%3E
  node
-s
 ;%0A  
@@ -1225,24 +1225,37 @@
  try %7B%0A     
+ List%3CRTNode%3E
  nodes = ser
@@ -1276,16 +1276,48 @@
 (ctxt);%0A
+      this.node = nodes.get(0);%0A
     %7D ca
@@ -1585,42 +1585,8 @@
 ) %7B%0A
-    for (RTNode node : nodes) %7B%0A  
     
@@ -1612,22 +1612,16 @@
 alues);%0A
-    %7D%0A
   %7D%0A%0A  @
@@ -1717,42 +1717,8 @@
 ) %7B%0A
-    for (RTNode node : nodes) %7B%0A  
     
@@ -1733,22 +1733,16 @@
 anup();%0A
-    %7D%0A
     ctxt

Make it clear that CrunchReducer only ever uses one child RTNode