make isReady only apply to transactions that haven't been emitted before

Nathan Marz · Nathan Marz · commit 7ae40de6d870 · 2012-03-28T00:36:30.000-07:00
diff --git a/src/jvm/backtype/storm/transactional/TransactionalSpoutCoordinator.java b/src/jvm/backtype/storm/transactional/TransactionalSpoutCoordinator.java
@@ -122,10 +122,11 @@ private void sync() {
         }
         
         try {
-            if(_coordinator.isReady() && _activeTx.size() < _maxTransactionActive) {
+            if(_activeTx.size() < _maxTransactionActive) {
                 BigInteger curr = _currTransaction;
                 for(int i=0; i<_maxTransactionActive; i++) {
-                    if(!_activeTx.containsKey(curr)) {
+                    if((_coordinatorState.hasCache(curr) || _coordinator.isReady())
+                            && !_activeTx.containsKey(curr)) {
                         TransactionAttempt attempt = new TransactionAttempt(curr, Utils.randomLong());
                         Object state = _coordinatorState.getState(curr, _initializer);
                         _activeTx.put(curr, new TransactionStatus(attempt));
diff --git a/src/jvm/backtype/storm/transactional/state/RotatingTransactionalState.java b/src/jvm/backtype/storm/transactional/state/RotatingTransactionalState.java
@@ -88,6 +88,10 @@ public Object getState(BigInteger txid, StateInitializer init) {
         }
         return _curr.get(txid);
     }
+    
+    public boolean hasCache(BigInteger txid) {
+        return _curr.containsKey(txid);
+    }
        
     /**
      * Returns null if it was created, the value otherwise.
diff --git a/test/clj/backtype/storm/transactional_test.clj b/test/clj/backtype/storm/transactional_test.clj
@@ -24,7 +24,7 @@
        nil)
      (getCoordinator [this conf context]
        (reify ITransactionalSpout$Coordinator
-         (isReady [this] true)
+         (isReady [this] (not (nil? @atom)))
          (initializeTransaction [this txid prevMetadata]
            @atom )
          (close [this]
@@ -115,12 +115,17 @@
         (.ack coordinator commit-id)
         (bind commit-id (get-commit emit-capture))
         (verify-and-reset! {COMMIT-STREAM [[2]] BATCH-STREAM [[5 12]]} emit-capture)
+        (reset! coordinator-state nil)
         (.ack coordinator commit-id)
-        (verify-and-reset! {BATCH-STREAM [[6 12]]} emit-capture)
+        (verify-and-reset! {} emit-capture)
 
         (.fail coordinator (nth attempts 1))
         (bind attempts (get-attempts emit-capture BATCH-STREAM))
-        (verify-and-reset! {BATCH-STREAM [[3 10] [4 10] [5 12] [6 12]]} emit-capture)
+        (verify-and-reset! {BATCH-STREAM [[3 10] [4 10] [5 12]]} emit-capture)
+
+        (reset! coordinator-state 12)
+        (.nextTuple coordinator)
+        (verify-and-reset! {BATCH-STREAM [[6 12]]} emit-capture)
 
         (.ack coordinator (first attempts))
         (bind commit-id (get-commit emit-capture))

Original file line number	Diff line number	Diff line change
`@@ -88,6 +88,10 @@ public Object getState(BigInteger txid, StateInitializer init) {`
`88`	`88`	`}`
`89`	`89`	`return _curr.get(txid);`
`90`	`90`	`}`
	`91`	`+`
	`92`	`+ public boolean hasCache(BigInteger txid) {`
	`93`	`+ return _curr.containsKey(txid);`
	`94`	`+ }`
`91`	`95`
`92`	`96`	`/**`
`93`	`97`	`* Returns null if it was created, the value otherwise.`