Merge branch 'master' into 0.8.2

Nathan Marz · Nathan Marz · commit cfdb01b5f742 · 2012-12-12T15:04:51.000-08:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -29,9 +29,11 @@
  * Added MultiScheme interface (thanks sritchie)
  * Added MockTridentTuple for testing (thanks emblem)
  * Updated Trident Debug filter to take in an identifier to use when logging (thanks emblem)
+ * Bug fix: Fix for bug that could cause topology to hang when ZMQ blocks sending to a worker that got reassigned
  * Bug fix: Fix deadlock bug due to variant of dining philosophers problem. Spouts now use an overflow buffer to prevent blocking and guarantee that it can consume the incoming queue of acks/fails.
  * Bug fix: Fix race condition in supervisor that would lead to supervisor continuously crashing due to not finding "stormconf.ser" file for an already killed topology
  * Bug fix: bin/storm script now displays a helpful error message when an invalid command is specified
+ * Bug fix: fixed NPE when emitting during emit method of Aggregator
   
 ## 0.8.1
 
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`	`}`