AMBARI-3469. Allow reduce tasks of any locality. (Billie Rinaldi via mahadev)

ambari-server/src/main/java/org/apache/ambari/eventdb/webservice/WorkflowJsonService.java

@@ -374,7 +374,7 @@ public class WorkflowJsonService {
     data.setMapNodeLocal(processExactLocalityData(taskAttempts, "MAP", "NODE_LOCAL", io));
     data.setMapRackLocal(processExactLocalityData(taskAttempts, "MAP", "RACK_LOCAL", io));
     data.setMapOffSwitch(processExactLocalityData(taskAttempts, "MAP", "OFF_SWITCH", io));
-    data.setReduceOffSwitch(processExactLocalityData(taskAttempts, "REDUCE", "OFF_SWITCH", io));
+    data.setReduceOffSwitch(processExactLocalityData(taskAttempts, "REDUCE", null, io));
     setRValues(data.getMapNodeLocal(), minr, maxr, io.max);
     setRValues(data.getMapRackLocal(), minr, maxr, io.max);
     setRValues(data.getMapOffSwitch(), minr, maxr, io.max);
@@ -462,7 +462,7 @@ public class WorkflowJsonService {
   private static List<DataPoint> processExactLocalityData(List<TaskAttempt> taskAttempts, String taskType, String locality, MinMax io) {
     List<DataPoint> data = new ArrayList<DataPoint>();
     for (TaskAttempt taskAttempt : taskAttempts) {
-      if (taskType.equals(taskAttempt.getTaskType()) && locality.equals(taskAttempt.getLocality())) {
+      if (taskType.equals(taskAttempt.getTaskType()) && (locality == null || locality.equals(taskAttempt.getLocality()))) {
         DataPoint point = new DataPoint();
         point.setX(taskAttempt.getStartTime());
         point.setY(taskAttempt.getFinishTime() - taskAttempt.getStartTime());