fix memorymapstate and lrumemorymapstate to avoid concurrent access t…

…o same map, update trident integration test of getting all tuples to properly do broadcast
nathanmarz · May 29, 2013 · dc8a3d1 · dc8a3d1
1 parent 45d0fae
commit dc8a3d1
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 2 deletions.
diff --git a/src/jvm/storm/trident/testing/LRUMemoryMapState.java b/src/jvm/storm/trident/testing/LRUMemoryMapState.java
@@ -72,7 +72,7 @@ public Factory(int maxSize) {
 
         @Override
         public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
-            return new LRUMemoryMapState(_maxSize, _id);
+            return new LRUMemoryMapState(_maxSize, _id + partitionIndex);
         }
     }
 

diff --git a/src/jvm/storm/trident/testing/MemoryMapState.java b/src/jvm/storm/trident/testing/MemoryMapState.java
@@ -69,7 +69,7 @@ public Factory() {
 
         @Override
         public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
-            return new MemoryMapState(_id);
+            return new MemoryMapState(_id + partitionIndex);
         }
     }
 

diff --git a/test/clj/storm/trident/integration_test.clj b/test/clj/storm/trident/integration_test.clj
@@ -26,6 +26,7 @@
               ))       
         (-> topo
             (.newDRPCStream "all-tuples" drpc)
+            (.broadcast)
             (.stateQuery word-counts (fields "args") (TupleCollectionGet.) (fields "word" "count"))
             (.project (fields "word" "count")))
         (with-topology [cluster topo]