twitter/summingbird — pull request #217 by sritchie (2013-09-17 06:21:26+00:00)
Title: Fix Storm Planner Bug

Verified the fix with some println debugging plus this:

```scala
object Test {
import com.twitter.storehaus._
import com.twitter.storehaus.algebra.MergeableStore
import com.twitter.summingbird.storm._
import com.twitter.summingbird._
import com.twitter.summingbird.batch.{BatchID, Batcher}
import com.twitter.tormenta.spout.Spout

implicit val te = TimeExtractor[Int](i => i.toLong)
implicit val batcher = Batcher.ofHours(1)
val node = Storm.source(Spout.fromTraversable(Seq(1,2,3))).flatMap { i => List(i, i, i) }

val newNode = node.flatMap(i => Some(i)).write(() => _ => com.twitter.util.Future.Unit)

try {
Storm.local("face").plan {
node.either(newNode).flatMap {
case Left(t) => Some(t)
case Right(r) => None
}.map(_ -> 1L).sumByKey {
MergeableStoreSupplier.from {
MergeableStore.fromStore[(Int, BatchID), Long](new JMapStore())
}
}
}
} catch {
case _ => null
}
}
```