在apache flink中拆分和加入流

"我想我有一个相当不规范的用例。我想使用以下filter函数将源流分割为多个流：

val dataStream:DataStream[MyEvent] = ...
val s1 = dataStream.filter(...).map(...)
val s2 = dataStream.filter(...).map(...)
我还有一个时间戳提取器（传入的事件将在XML中附加一个时间戳）：

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
...

dataStream.assignTimestampsAndWatermarks(new MyTimestampExtractor)
...

class MyTimestampExtractor extends AssignerWithPunctuatedWatermarks[Elem]
{
override def checkAndGetNextWatermark(lastElement:Elem, extractedTimestamp:Long):Watermark = new Watermark(extractedTimestamp)
override def extractTimestamp(element:Elem, previousElementTimestamp:Long):Long = XmlOperations.getDateTime(element, ""@timestamp"").getMillis
}
我之所以选择这种方法，并不是简单地做一个单一的流（val s = dataStream.filter(...).map(...).filter(...).map(...)），因为我想构建一个分割/组合任意流的网络（例如s1 + s2-> c1，s1 + s3-> c2，c2 + s4-> c3， ...）

现在，当通过上面的例子发送事件时，可能是事件E1在s1和s2中都结束了。根据我的理解，这意味着将同一事件E1作为第一个实例放入s1（E1a），并将第二个实例放入s2（E1b）。

所以我现在要做的是将E1a和E1b重新组合成一个组合E1，它类似于E1，它既是s1和s2的转换。

我试过了：
val c1 = s1.join(s2)
.where(_.key).equalTo(_.key)
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.apply((e1a, e2b) => { printf(""Got e1a and e1b""); e1a })
然而，似乎事件永远不会达到应用功能，我无法找到原因。

我的例子有什么问题？我的这种流网络的方法/想法是否会起作用？"