error checking on ComboList, and fix bug in Aggregator so that it sets the correct group during aggregate

Nathan Marz · Nathan Marz · commit fd201eb15a1d · 2012-12-12T14:58:33.000-08:00
diff --git a/src/jvm/storm/trident/operation/impl/GroupedAggregator.java b/src/jvm/storm/trident/operation/impl/GroupedAggregator.java
@@ -56,6 +56,7 @@ public void aggregate(Object[] arr, TridentTuple tuple, TridentCollector collect
         } else {
             curr = val.get(group);
         }
+        groupColl.currGroup = group;
         _agg.aggregate(curr, input, groupColl);
         
     }
diff --git a/src/jvm/storm/trident/tuple/ComboList.java b/src/jvm/storm/trident/tuple/ComboList.java
@@ -3,15 +3,16 @@
 import java.io.Serializable;
 import java.util.AbstractList;
 import java.util.List;
+import org.apache.commons.lang.builder.ToStringBuilder;
 
 
 public class ComboList extends AbstractList<Object> {    
     public static class Factory implements Serializable {
         Pointer[] index;
-        int numLists;
+        int[] sizes;
         
         public Factory(int... sizes) {
-            numLists = sizes.length;
+            this.sizes = sizes;
             int total = 0;
             for(int size: sizes) {
                 total+=size;
@@ -29,8 +30,14 @@ public Factory(int... sizes) {
         }
         
         public ComboList create(List[] delegates) {
-            if(delegates.length!=numLists) {
-                throw new RuntimeException("Expected " + numLists + " lists, but instead got " + delegates.length + " lists");
+            if(delegates.length!=sizes.length) {
+                throw new RuntimeException("Expected " + sizes.length + " lists, but instead got " + delegates.length + " lists");
+            }
+            for(int i=0; i<delegates.length; i++) {
+                List l = delegates[i];
+                if(l==null || l.size() != sizes[i]) {
+                    throw new RuntimeException("Got unexpected delegates to ComboList: " + ToStringBuilder.reflectionToString(delegates));
+                }
             }
             return new ComboList(delegates, index);
         }

Original file line number	Diff line number	Diff line change
`@@ -56,6 +56,7 @@ public void aggregate(Object[] arr, TridentTuple tuple, TridentCollector collect`
`56`	`56`	`} else {`
`57`	`57`	`curr = val.get(group);`
`58`	`58`	`}`
	`59`	`+ groupColl.currGroup = group;`
`59`	`60`	`_agg.aggregate(curr, input, groupColl);`
`60`	`61`
`61`	`62`	`}`