fix bug in TransactionalSpoutBatchExecutor where cleanup could happen…

… before the new metadata is saved

src/jvm/backtype/storm/transactional/TransactionalSpoutBatchExecutor.java

@@ -34,13 +34,13 @@ public void execute(Tuple input) {
         try {
             _emitter.emitBatch(attempt, input.getValue(1), _collector);
             _collector.ack(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));
         } catch(FailedException e) {
             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));
     }
 
     @Override