Storm platform fails in case of `sumByKey` in one branch and `flatMap` in another with `flatMap`ed source fails in runtime

[ttim]

Producer

val source = spoutSource.flatMap(e => List(e))
  source.sumByKey(store1).also(
    source.flatMap(branchFlatMap).sumByKey(store2)
  )

fails with java.lang.ClassCastException: java.lang.Integer cannot be cast to scala.Tuple2 exception in runtime.

[johnynek]

would be nice to fix this. Scary that something like this lasted this long.