fix cluster mode for MemoryTransactionalSpout (don't update finishedStatuses in cluster mode)

Nathan Marz · Nathan Marz · commit eacd41cf0743 · 2012-05-24T15:51:48.000-07:00
diff --git a/src/jvm/backtype/storm/testing/MemoryTransactionalSpout.java b/src/jvm/backtype/storm/testing/MemoryTransactionalSpout.java
@@ -97,7 +97,11 @@ public MemoryTransactionalSpoutMeta emitPartitionBatchNew(TransactionAttempt tx,
                 int curr = Utils.get(_emptyPartitions, partition, 0) + 1;
                 _emptyPartitions.put(partition, curr);
                 if(curr > _maxSpoutPending) {
-                    getFinishedStatuses().put(partition, true);
+                    Map<Integer, Boolean> finishedStatuses = getFinishedStatuses();
+                    // will be null in remote mode
+                    if(finishedStatuses!=null) {
+                        finishedStatuses.put(partition, true);
+                    }
                 }
             }
             return ret;   

Original file line number	Diff line number	Diff line change
`@@ -97,7 +97,11 @@ public MemoryTransactionalSpoutMeta emitPartitionBatchNew(TransactionAttempt tx,`
`97`	`97`	`int curr = Utils.get(_emptyPartitions, partition, 0) + 1;`
`98`	`98`	`_emptyPartitions.put(partition, curr);`
`99`	`99`	`if(curr > _maxSpoutPending) {`
`100`		`- getFinishedStatuses().put(partition, true);`
	`100`	`+ Map<Integer, Boolean> finishedStatuses = getFinishedStatuses();`
	`101`	`+ // will be null in remote mode`
	`102`	`+ if(finishedStatuses!=null) {`
	`103`	`+ finishedStatuses.put(partition, true);`
	`104`	`+ }`
`101`	`105`	`}`
`102`	`106`	`}`
`103`	`107`	`return ret;`