Fix Storm Planner Bug

[sritchie]

Verified the fix with some println debugging plus this:

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
  }
}

[johnynek]

we could .recover here to make sure a write doesn't drop the downstream.

Is that the right thing or let it fail, and see the restarts in nimbus?

[sritchie]

I say we rescue. We need to do the same thing after service joins, I think. To do this right, I think we need default handlers all over.