fix race condition in coordinatedbolt

Nathan Marz · Nathan Marz · commit ff2e570b8067 · 2012-01-25T12:27:11.000-08:00
diff --git a/src/jvm/backtype/storm/coordination/CoordinatedBolt.java b/src/jvm/backtype/storm/coordination/CoordinatedBolt.java
@@ -136,6 +136,7 @@ public static class TrackingInfo {
         int failedTuples = 0;
         Map<Integer, Integer> taskEmittedTuples = new HashMap<Integer, Integer>();
         boolean receivedId = false;
+        boolean finished = false;
         
         @Override
         public String toString() {
@@ -232,6 +233,7 @@ private void checkFinishId(Tuple tup) {
                     int numTuples = get(track.taskEmittedTuples, task, 0);
                     _collector.emitDirect(task, Constants.COORDINATED_STREAM_ID, tup, new Values(id, numTuples));
                 }
+                track.finished = true;
                 _tracked.remove(id);
             }
         }
@@ -308,7 +310,11 @@ private class TimeoutItems implements TimeCacheMap.ExpiredCallback<Object, Track
         @Override
         public void expire(Object id, TrackingInfo val) {
             synchronized(_tracked) {
-                ((TimeoutCallback) _delegate).timeoutId(id);
+                // make sure we don't time out something that has been finished. the combination of 
+                // the flag and the lock ensure this
+                if(!val.finished) {
+                    ((TimeoutCallback) _delegate).timeoutId(id);
+                }
             }
         }
     } 
diff --git a/src/jvm/backtype/storm/transactional/TransactionalSpoutBatchExecutor.java b/src/jvm/backtype/storm/transactional/TransactionalSpoutBatchExecutor.java
@@ -38,6 +38,8 @@ public void execute(Tuple input) {
             LOG.warn("Failed to emit batch for transaction", e);
             _collector.fail(input);
         }
+        // this is valid here because the batch has been successfully emitted, 
+        // so we can safely delete metadata for prior transactions
         _emitter.cleanupBefore((BigInteger) input.getValue(2));
     }
 

Original file line number	Diff line number	Diff line change
`@@ -136,6 +136,7 @@ public static class TrackingInfo {`
`136`	`136`	`int failedTuples = 0;`
`137`	`137`	`Map<Integer, Integer> taskEmittedTuples = new HashMap<Integer, Integer>();`
`138`	`138`	`boolean receivedId = false;`
	`139`	`+ boolean finished = false;`
`139`	`140`
`140`	`141`	`@Override`
`141`	`142`	`public String toString() {`
`@@ -232,6 +233,7 @@ private void checkFinishId(Tuple tup) {`
`232`	`233`	`int numTuples = get(track.taskEmittedTuples, task, 0);`
`233`	`234`	`_collector.emitDirect(task, Constants.COORDINATED_STREAM_ID, tup, new Values(id, numTuples));`
`234`	`235`	`}`
	`236`	`+ track.finished = true;`
`235`	`237`	`_tracked.remove(id);`
`236`	`238`	`}`
`237`	`239`	`}`
`@@ -308,7 +310,11 @@ private class TimeoutItems implements TimeCacheMap.ExpiredCallback<Object, Track`
`308`	`310`	`@Override`
`309`	`311`	`public void expire(Object id, TrackingInfo val) {`
`310`	`312`	`synchronized(_tracked) {`
`311`		`- ((TimeoutCallback) _delegate).timeoutId(id);`
	`313`	`+ // make sure we don't time out something that has been finished. the combination of`
	`314`	`+ // the flag and the lock ensure this`
	`315`	`+ if(!val.finished) {`
	`316`	`+ ((TimeoutCallback) _delegate).timeoutId(id);`
	`317`	`+ }`
`312`	`318`	`}`
`313`	`319`	`}`
`314`	`320`	`}`
Original file line number	Diff line number	Diff line change
`@@ -38,6 +38,8 @@ public void execute(Tuple input) {`
`38`	`38`	`LOG.warn("Failed to emit batch for transaction", e);`
`39`	`39`	`_collector.fail(input);`
`40`	`40`	`}`
	`41`	`+ // this is valid here because the batch has been successfully emitted,`
	`42`	`+ // so we can safely delete metadata for prior transactions`
`41`	`43`	`_emitter.cleanupBefore((BigInteger) input.getValue(2));`
`42`	`44`	`}`
`43`	`45`