fix memorymapstate and lrumemorymapstate to avoid concurrent access to same map, update trident integration test of getting all tuples to properly do broadcast

nathanmarz · nathanmarz · commit dc8a3d10c38b · 2013-05-29T13:43:21.000-07:00
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]

Original file line number	Diff line number	Diff line change
`@@ -72,7 +72,7 @@ public Factory(int maxSize) {`
`72`	`72`
`73`	`73`	`@Override`
`74`	`74`	`public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {`
`75`		`- return new LRUMemoryMapState(_maxSize, _id);`
	`75`	`+ return new LRUMemoryMapState(_maxSize, _id + partitionIndex);`
`76`	`76`	`}`
`77`	`77`	`}`
`78`	`78`
Original file line number	Diff line number	Diff line change
`@@ -69,7 +69,7 @@ public Factory() {`
`69`	`69`
`70`	`70`	`@Override`
`71`	`71`	`public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {`
`72`		`- return new MemoryMapState(_id);`
	`72`	`+ return new MemoryMapState(_id + partitionIndex);`
`73`	`73`	`}`
`74`	`74`	`}`
`75`	`75`